using Glpt.Common;
using Glpt.Data;
using SocialExplorer.IO.FastDBF;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.IO;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Glpt.Task
{
    /// <summary>
    /// 数据交换任务
    /// </summary>
    public class ExchDataTask
    {
        private BackgroundWorker bwt = null;
        private const string ATTR_PARAM_VALUE = "@PARAM_VALUE";
        private const string ATTR_SOURCE_TABLE_NAME = "@SOURCE_TABLE_NAME";
        private const string ATTR_TARGET_TABLE_NAME = "@TARGET_TABLE_NAME";
        private const string ATTR_SOURCE_DB_NO = "@SOURCE_DB_NO";
        private const string ATTR_TARGET_DB_NO = "@TARGET_DB_NO";
        private const string ATTR_NODE_ID = "@NODE_ID";
        private string new_function_id = "";
        private Dictionary<string, FileCombineInfo> exp_filelist = new Dictionary<string, FileCombineInfo>();//导出文件清单
        private Object exp_filelist_Lock = new Object();
        private Exception taskException = null;//错误信息
        private string ERROR_MSG = "";//错误信息
        private string FILE_TIME = "";//文件时间
        private long sno = 0;//流水号
        public bool IsSqlLog = false;//是否记录SQL日志
        private bool IsFileLog = false;//是否记录文件日志
        public bool IsWeb = false;//是否WEB调用
        public DataTable MemDataTable = null;
        private Object task_Lock = new Object();
        private Object schema_Lock = new Object();
        private Object sql_Lock = new Object();
        private static DataTable NetPathMapTable = null;
        private int WRITE_THREADCOUNT = 0;
        private int MAXERRORS = 0;
        private int TIMEOUT = -1;
        private string SPLIT_COLNAME = "";
        private bool WRITE_LOG = false;
        private Object ExecuteInfo_Lock = new Object();
        private Dictionary<DbItem, ExecuteInfo> ExecuteInfoList = new Dictionary<DbItem, ExecuteInfo>();//执行信息
        private int ExecuteTotalCount = 0;
        private Object ForCount_Lock = new Object();
        private Dictionary<string, int> ForCount = new Dictionary<string, int>();//循环次数
        private Dictionary<DbItem, DataTable> TargetTableInfoList = new Dictionary<DbItem, DataTable>();//对方数据库表结构
        private Object AfterSQL_Lock = new Object();
        private BaseInterFace zkbif = null;
        private string SOURCE_DB_LIST = "";
        private bool is_change_over = false;//是否切换到备库
        /// <summary>
        /// 初始化类的新实例
        /// </summary>
        public ExchDataTask()
        {
        }
        /// <summary>
        /// 参数设置
        /// </summary>
        /// <param name="function_id">功能号</param>
        public void SetParams(string function_id)
        {
            new_function_id = function_id;
        }
        /// <summary>
        /// 测试接口
        /// </summary>
        /// <param name="function_id">功能号</param>
        /// <param name="tradedate">业务日期</param>
        public int SetParams(string function_id, string tradedate)
        {
            return SetParams(function_id, tradedate, "");
        }
        /// <summary>
        /// 参数设置
        /// </summary>
        /// <param name="function_id">功能编号</param>
        /// <param name="tradedate">业务日期</param>
        /// <param name="exec_value">参数</param>
        public int SetParams(string function_id, string tradedate, string exec_value)
        {
            return SetParams(0, function_id, "", tradedate, exec_value);
        }
        /// <summary>
        /// 参数设置
        /// </summary>
        /// <param name="db_no">数据库编号</param>
        /// <param name="function_id">功能编号</param>
        /// <param name="task_id">任务编号</param>
        /// <param name="tradedate">业务日期</param>
        public int SetParams(int db_no, string function_id, string task_id, string tradedate)
        {
            return SetParams(db_no, function_id, task_id, tradedate, "");
        }
        /// <summary>
        /// 参数设置
        /// </summary>
        /// <param name="db_no">数据库编号</param>
        /// <param name="function_id">功能编号</param>
        /// <param name="task_id">任务编号</param>
        /// <param name="tradedate">业务日期</param>
        /// <param name="exec_value">参数</param>
        public int SetParams(int db_no, string function_id, string task_id, string tradedate, string exec_value)
        {
            if (new_function_id.Length > 0)
            {
                if (function_id.Equals(new_function_id))//功能号相同，切换到进程
                {
                    string forwardexe = AbstractInterFace.StartupPath + "ExchData.exe";
                    Dictionary<string, string> param = new Dictionary<string, string>();
                    param.Add("LoginID", BaseInterFace.LoginID);
                    param.Add("db_no", db_no.ToString());
                    param.Add("function_id", function_id);
                    param.Add("task_id", task_id);
                    param.Add("tradedate", tradedate);
                    param.Add("exec_value", exec_value);
                    string sval = "";
                    string SepChar = "|`|";
                    foreach (string key in param.Keys)
                    {
                        if (param[key].Length == 0) continue;
                        if (sval.Length > 0) sval += SepChar;
                        sval += key + SepChar + param[key];
                    }
                    string arguments = Convert.ToBase64String(DataFuns.DefaultEncoding.GetBytes(sval));
                    string msg = InvokeDynamic.InvokeEXE(forwardexe, arguments).TrimEnd();
                    string msg_ret = "执行成功返回值：";
                    int msg_idx = msg.IndexOf(msg_ret);
                    if (msg_idx >= 0)
                    {
                        int iret = 0;
                        if (int.TryParse(msg.Substring(msg_idx + msg_ret.Length), out iret))
                            return iret;
                        else
                            throw new Exception(msg);
                    }
                    msg_ret = "执行失败返回值：";
                    msg_idx = msg.IndexOf(msg_ret);
                    if (msg_idx >= 0)
                        throw new Exception(msg.Substring(msg_idx + msg_ret.Length));
                    else
                        throw new Exception(msg);
                }
                else
                    function_id = new_function_id;//切换功能号
            }
            Dictionary<string, string> exec_params = new Dictionary<string, string>();
            Dictionary<string, string> init_params = new Dictionary<string, string>();
            if (!Funcs.IsDate(tradedate))//参数不是日期
            {
                Dictionary<string, string> arr_params = DataFuns.SplitStringValue(tradedate, ',');
                foreach (string key in arr_params.Keys)
                {
                    CommFuns.AddParams(init_params, key, arr_params[key]);
                }
                List<string> arr_date = DataFuns.SplitString(tradedate, ',', true);
                tradedate = "";
                if (arr_date.Count >= 1)
                {
                    if (Funcs.IsDate(arr_date[0]))
                    {
                        tradedate = arr_date[0];
                        init_params.Remove(tradedate);
                    }
                }
                if (arr_date.Count >= 3)
                {
                    if (Funcs.IsDate(arr_date[1]))
                    {
                        string begin_date = arr_date[1];
                        init_params.Remove(begin_date);
                        CommFuns.AddParams(init_params, "BEGIN_DATE", begin_date);
                    }
                    if (Funcs.IsDate(arr_date[2]))
                    {
                        string end_date = arr_date[2];
                        init_params.Remove(end_date);
                        CommFuns.AddParams(init_params, "END_DATE", end_date);
                    }
                }
                if (tradedate.Length == 0) tradedate = DateTime.Now.ToString("yyyyMMdd");
            }
            foreach (string key in init_params.Keys)//把初始参数加回去
            {
                CommFuns.AddParams(exec_params, key.Substring(1), init_params[key]);
            }
            string exch_id_name = "";
            string param_value = "";
            int idx = exec_value.IndexOf(":");
            if (idx >= 0)
            {
                exch_id_name = exec_value.Substring(0, idx);
                param_value = exec_value.Substring(idx + 1);
            }
            else
                exch_id_name = exec_value;
            zkbif = new BaseInterFace(db_no);
            string sql = "SELECT * FROM TB_EXCH_LIST WHERE TASK_FUNCTION_ID=" + function_id + " AND ENABLE_FLAG='1'";
            if (exch_id_name.Length > 0)
            {
                if (Funcs.IsInt(exch_id_name))
                    sql += " AND EXCH_ID=" + exch_id_name;
                else if (exch_id_name.StartsWith("+"))
                    sql += " AND TARGET_TABLE_NAME LIKE '" + exch_id_name.Substring(1) + "'";
                else
                    sql += " AND SOURCE_TABLE_NAME LIKE '" + exch_id_name + "'";
            }
            else if (task_id.Length > 0)
                sql += " AND EXCH_ID NOT IN (SELECT SQLCMD_ID FROM TB_SQLCMD_LOG WHERE TASK_ID=" + task_id + ")";
            DataTable dt_list = zkbif.ExecSelect(sql);
            int icount = dt_list.Rows.Count;
            if (IsWeb && NetPathMapTable == null)
            {
                sql = "SELECT * FROM TB_NET_PATH_MAP";
                NetPathMapTable = zkbif.ExecSelect(sql);
            }
            foreach (DataRow dr in dt_list.Rows)
            {
                if (bwt != null && bwt.CancellationPending)
                    throw new Exception("线程被取消");
                if (ReturnMemDataTable && MemDataTable != null)
                    break;
                TargetTableInfoList.Clear();
                IsFileLog = false;
                WRITE_THREADCOUNT = 0;
                MAXERRORS = 0;
                TIMEOUT = -1;
                SPLIT_COLNAME = "";
                string exch_id = dr["EXCH_ID"].ToString();
                WRITE_LOG = (dr["WRITE_LOG"].ToString().Equals("1"));
                SOURCE_DB_LIST = dr["SOURCE_DB_LIST"].ToString();
                DbItemCollection db_source_list = GetDbItemCollection(zkbif, SOURCE_DB_LIST, dr["SOURCE_DB_NO"].ToString());//源数据库
                DbItemCollection db_target_list = GetDbItemCollection(zkbif, dr["TARGET_DB_LIST"].ToString(), dr["TARGET_DB_NO"].ToString());//目标数据库
                DateTime begin_time = DateTime.Now;
                string file_msg = "";
                int ido = 0;
                List<string> op_exp_list = new List<string>();
                List<DataTaskParam> PList = new List<DataTaskParam>();
                ExecuteInfoList.Clear();
                ExecuteTotalCount = 0;
                ForCount.Clear();
                lock (ForCount_Lock) { if (!ForCount.ContainsKey("1")) ForCount.Add("1", db_source_list.Count); }
                foreach (DbItem db_source in db_source_list)
                {
                    if (bwt != null && bwt.CancellationPending)
                        throw new Exception("线程被取消");
                    if (ReturnCreateTableSQL && CreateTableSQL.Length > 0)
                        break;
                    bool imp_file = db_source.DB_TYPE.Equals(DataBaseType.Dbf);
                    string SOURCE_TABLE_NAME = dr["SOURCE_TABLE_NAME"].ToString();
                    string TARGET_TABLE_NAME = dr["TARGET_TABLE_NAME"].ToString();
                    string DATE_EXP = dr["DATE_EXP"].ToString().Replace(ATTR_PARAM_VALUE, param_value);
                    if (DATE_EXP.Length > 0)//有日期函数
                    {
                        DATE_EXP = Funcs.ReplaceWord(DATE_EXP.Replace(ATTR_SOURCE_TABLE_NAME, SOURCE_TABLE_NAME).Replace(ATTR_TARGET_TABLE_NAME, TARGET_TABLE_NAME), tradedate);
                        List<string> date_exp_list = DataFuns.SplitString(DATE_EXP, ';', false);
                        bool isret = false;
                        foreach (string val in date_exp_list)
                        {
                            string sql_val = val;
                            foreach (string key in exec_params.Keys)
                            {
                                if (exec_params[key].Length > 0)
                                    sql_val = sql_val.Replace(key, exec_params[key]);
                            }
                            DataTable dt = CommFuns.GetDynamicFunc(zkbif, sql_val, "BUSI_DATE");//获取日期函数值
                            if (dt.Rows.Count > 0)
                                CommFuns.AddParams(exec_params, dt);
                            else
                            {
                                isret = true;
                                break;
                            }
                        }
                        if (isret)
                            continue;
                    }
                    foreach (string key in init_params.Keys)//把初始参数加回去
                    {
                        CommFuns.AddParams(exec_params, key.Substring(1), init_params[key]);
                    }
                    if (exec_params.ContainsKey("@BEGIN_DATE") && exec_params["@BEGIN_DATE"].Length == 8 && !exec_params.ContainsKey("@C_BEGIN_DATE"))
                        CommFuns.AddParams(exec_params, "C_BEGIN_DATE", Funcs.StrToDataTime(exec_params["@BEGIN_DATE"]).ToString("yyyy-MM-dd"));
                    if (exec_params.ContainsKey("@END_DATE") && exec_params["@END_DATE"].Length == 8 && !exec_params.ContainsKey("@C_END_DATE"))
                        CommFuns.AddParams(exec_params, "C_END_DATE", Funcs.StrToDataTime(exec_params["@END_DATE"]).ToString("yyyy-MM-dd"));
                    string BUSI_DATE = tradedate;
                    if (exec_params.ContainsKey("@BUSI_DATE") && exec_params["@BUSI_DATE"].Length > 0) BUSI_DATE = exec_params["@BUSI_DATE"];
                    if (imp_file)
                    {
                        if (exec_params.ContainsKey(ATTR_SOURCE_TABLE_NAME))
                        {
                            if (exec_params[ATTR_SOURCE_TABLE_NAME].Length > 0)
                                SOURCE_TABLE_NAME = exec_params[ATTR_SOURCE_TABLE_NAME].Replace(CommFuns.StringSplitChar.ToString(), ",");
                            else
                                continue;
                        }
                        if (exec_params.ContainsKey(ATTR_TARGET_TABLE_NAME))
                        {
                            if (exec_params[ATTR_TARGET_TABLE_NAME].Length > 0)
                                TARGET_TABLE_NAME = exec_params[ATTR_TARGET_TABLE_NAME].Replace(CommFuns.StringSplitChar.ToString(), ",");
                            else
                                continue;
                        }
                    }
                    SOURCE_TABLE_NAME = Funcs.ReplaceWord(SOURCE_TABLE_NAME.Replace(ATTR_PARAM_VALUE, param_value), BUSI_DATE);
                    TARGET_TABLE_NAME = Funcs.ReplaceWord(TARGET_TABLE_NAME.Replace(ATTR_PARAM_VALUE, param_value), BUSI_DATE);
                    string SQL_EXP = Funcs.ReplaceWord(dr["SQL_EXP"].ToString(), BUSI_DATE).Replace(ATTR_PARAM_VALUE, param_value).Trim();//查询数据的SQL语句
                    string COND_EXP = Funcs.ReplaceWord(dr["COND_EXP"].ToString().Trim(), BUSI_DATE);//条件函数                    
                    string OP_EXP = Funcs.ReplaceWord(dr["OP_EXP"].ToString(), BUSI_DATE).Replace(ATTR_PARAM_VALUE, param_value).Trim();//清除数据的SQL语句
                    DataFileType dft = GetDataFileType("", SQL_EXP);
                    foreach (string param_name in exec_params.Keys)
                    {
                        string val = Funcs.ReplaceWord(exec_params[param_name], BUSI_DATE).Replace(CommFuns.StringSplitChar.ToString(), ",");
                        COND_EXP = COND_EXP.Replace(param_name, val);
                        OP_EXP = OP_EXP.Replace(param_name, val);
                        SOURCE_TABLE_NAME = SOURCE_TABLE_NAME.Replace(param_name, val);
                        TARGET_TABLE_NAME = TARGET_TABLE_NAME.Replace(param_name, val);
                    }
                    op_exp_list = DataFuns.SplitString(OP_EXP, ';', false);
                    if (op_exp_list.Count > 0)
                    {
                        OP_EXP = op_exp_list[0];//第一个值是前置SQL
                        op_exp_list.RemoveAt(0);
                    }
                    if (TARGET_TABLE_NAME.Equals("#"))//目标表是#表示返回内存表数据
                        ReturnMemDataTable = true;
                    else if (TARGET_TABLE_NAME.StartsWith("#") && TARGET_TABLE_NAME.Length > 1 && !ReturnMemDataTable)//临时表
                    {
                        if (op_exp_list.Count == 0)//没有配置时默认自动结转到目标表
                        {
                            OP_EXP = "TRUNCATE TABLE " + TARGET_TABLE_NAME.Substring(1);
                            op_exp_list.Add("INSERT INTO " + TARGET_TABLE_NAME.Substring(1) + " SELECT * FROM " + TARGET_TABLE_NAME);
                        }
                    }
                    if (COND_EXP.Length > 0)//有条件函数
                    {
                        if (COND_EXP.Equals("CHECK_FILE_TIME"))
                        {
                            if (imp_file && task_id.Length > 0)
                            {
                                string filename_source = ReplaceNetPath(Funcs.ReplaceWord(Path.Combine(db_source.DB_IP, db_source.DB_NAME, SOURCE_TABLE_NAME), BUSI_DATE));
                                FileInfo file = new FileInfo(filename_source);
                                if (file.Exists)
                                {
                                    FILE_TIME = file.LastWriteTime.ToString("yyyy-MM-dd HH:mm:ss");
                                    DataTable dtlog = zkbif.ExecSelect("SELECT LOG_MESSAGE FROM TB_SQLCMD_LOG WHERE TASK_ID=" + task_id + " AND SQLCMD_ID=" + exch_id);
                                    if (dtlog.Rows.Count > 0)
                                    {
                                        string last_time = dtlog.Rows[0][0].ToString().Substring(0, 19);
                                        if (Funcs.IsDate(last_time) && FILE_TIME.CompareTo(last_time) <= 0)//文件时间没有变大
                                            continue;
                                    }
                                }
                                else
                                    continue;
                            }
                        }
                        else if (!CommFuns.GetDynamicFunc(zkbif, COND_EXP, "").Rows[0][0].ToString().Equals("1"))//获取条件函数的结果值
                            continue;
                    }
                    ido++;
                    dft = DataFileType.DATABASE;
                    int BATCHSIZE = 0;
                    int THREADCOUNT = 0;
                    List<string> filename_source_list = new List<string>();
                    List<string> sql_exp_list = new List<string>();
                    bool ALLOWNOTEXIST = false;
                    bool EXECBACKSQL = false;
                    if (imp_file)//数据源是文件类型
                    {
                        sql_exp_list.Add(SQL_EXP);
                        Dictionary<string, string> param = DataFuns.SplitStringValue(SQL_EXP, ',');
                        GetParamDictionary(task_id, param, ref BATCHSIZE, ref THREADCOUNT, ref WRITE_THREADCOUNT, ref TIMEOUT, ref MAXERRORS, ref IsFileLog, ref SPLIT_COLNAME);
                        if (param.ContainsKey("ALLOWNOTEXIST"))
                        {
                            if (param["ALLOWNOTEXIST"].Equals("2"))
                                EXECBACKSQL = true;
                            if (EXECBACKSQL || param["ALLOWNOTEXIST"].Equals("1"))
                                ALLOWNOTEXIST = true;
                        }
                        string[] source_name_list = SOURCE_TABLE_NAME.Split(new char[] { ',' }, StringSplitOptions.RemoveEmptyEntries);
                        foreach (string source_name in source_name_list)
                        {
                            if (FilePath.IsURL(source_name) || FilePath.IsFTP(source_name))
                                filename_source_list.Add(source_name);
                            else//检测文件是否存在
                            {
                                string filename_source = ReplaceNetPath(Funcs.ReplaceWord(Path.Combine(db_source.DB_IP, db_source.DB_NAME, source_name), BUSI_DATE));
                                idx = filename_source.LastIndexOf(DataFuns.DirSeparator);
                                if (idx > 0)
                                {
                                    string[] file_list = new string[0];
                                    if (Directory.Exists(filename_source.Substring(0, idx)))
                                    {
                                        if (filename_source.Substring(0, idx).EndsWith(":"))//根目录
                                            file_list = new string[] { filename_source };
                                        else
                                            file_list = Directory.GetFiles(filename_source.Substring(0, idx), filename_source.Substring(idx + 1), source_name.Contains("*") ? SearchOption.AllDirectories : SearchOption.TopDirectoryOnly);
                                    }
                                    if (file_list.Length > 0)
                                        filename_source_list.AddRange(file_list);
                                    else
                                    {
                                        if (file_msg.Length == 0)
                                            file_msg = "文件不存在：" + filename_source;
                                        else
                                            file_msg += "," + filename_source;
                                        if (!ALLOWNOTEXIST)
                                            throw new Exception(file_msg);
                                    }
                                }
                            }
                        }
                        if (filename_source_list.Count > 0)
                        {
                            string filename_source = filename_source_list[0];
                            AutoLoginNet(filename_source, db_source.USER_NAME, db_source.USER_PASSWORD);
                            dft = GetDataFileType(filename_source, SQL_EXP);
                        }
                        else if (ALLOWNOTEXIST)
                        {
                            if (!EXECBACKSQL) op_exp_list.Clear();
                            break;
                        }
                    }
                    if (db_source.DB_TYPE.Equals(DataBaseType.MongoDB) || db_source.DB_TYPE.Equals(DataBaseType.Kafka))//数据源是MongoDB、Kafka
                    {
                        sql_exp_list.Add(SQL_EXP);
                        Dictionary<string, string> param = DataFuns.SplitStringValue(SQL_EXP, ',');
                        GetParamDictionary(task_id, param, ref BATCHSIZE, ref THREADCOUNT, ref WRITE_THREADCOUNT, ref TIMEOUT, ref MAXERRORS, ref IsFileLog, ref SPLIT_COLNAME);
                    }
                    else if (dft.Equals(DataFileType.DATABASE))//数据源是数据库
                    {
                        sql_exp_list = DataFuns.SplitString(SQL_EXP, ';', false);//以分号分拆SQL语句
                        for (int i = sql_exp_list.Count - 1; i >= 0; i--)
                        {
                            sql_exp_list[i] = sql_exp_list[i].Trim();
                            if (sql_exp_list[i].Length == 0)
                            {
                                sql_exp_list.RemoveAt(i);
                                continue;
                            }
                            Dictionary<string, string> param = DataFuns.SplitStringValue(sql_exp_list[i], ',');//再以逗号分拆参数
                            bool ContainsKey = GetParamDictionary(task_id, param, ref BATCHSIZE, ref THREADCOUNT, ref WRITE_THREADCOUNT, ref TIMEOUT, ref MAXERRORS, ref IsFileLog, ref SPLIT_COLNAME);
                            if (param.ContainsKey("ALLOWNOTEXIST")) ALLOWNOTEXIST = param["ALLOWNOTEXIST"].Equals("1");
                            if (param.ContainsKey("NOT_UNION") && !exec_params.ContainsKey("NOT_UNION")) exec_params.Add("NOT_UNION", "1");
                            if (ContainsKey) sql_exp_list.RemoveAt(i);
                        }
                        if (sql_exp_list.Count == 0) sql_exp_list.Add("");
                    }
                    if (ReturnCreateTableSQL && THREADCOUNT > 1) THREADCOUNT = 1;//如果返回建表SQL语句，则禁用多线程
                    string TRAN_MODE = dr["TRAN_MODE"].ToString();//0数据流 1内存表 2INSERT语句 3文件
                    string MAP_MODE = dr["MAP_MODE"].ToString();//0按名称 1按顺序
                    string PAIR_MODE = dr["PAIR_MODE"].ToString();//字段配对方式
                    string PAIR_PARAMS = dr["PAIR_PARAMS"].ToString();//配对参数
                    string map_params = MAP_MODE + ',' + PAIR_MODE + ',' + PAIR_PARAMS;
                    if (IsFileLog) WriteLog(exch_id, "开始执行任务", true);
                    lock (ForCount_Lock) { if (!ForCount.ContainsKey("2")) ForCount.Add("2", sql_exp_list.Count); }
                    for (int j = 0; j < sql_exp_list.Count; j++)
                    {
                        if (bwt != null && bwt.CancellationPending)
                            throw new Exception("线程被取消");
                        if (ReturnCreateTableSQL && CreateTableSQL.Length > 0)//返回SQL建表脚本只执行一次
                            break;
                        string sql_exp2 = sql_exp_list[j];
                        if ((db_source_list.Count > 1 || sql_exp_list.Count > 1) && THREADCOUNT > 1)//多线程
                        {
                            DataTaskParam dtp = new DataTaskParam();
                            dtp.BUSI_DATE = BUSI_DATE;
                            dtp.DB_SOURCE = db_source;
                            dtp.IMP_FILE = imp_file;
                            dtp.FILENAME_SOURCE_LIST = filename_source_list;
                            dtp.SOURCE_TABLE_NAME = SOURCE_TABLE_NAME;
                            dtp.TARGET_TABLE_NAME = TARGET_TABLE_NAME;
                            dtp.SQL_EXP = sql_exp2;
                            dtp.OP_EXP = OP_EXP;
                            dtp.OP_EXP_LIST = op_exp_list;
                            dtp.TRAN_MODE = TRAN_MODE;
                            dtp.MAP_PARAMS = map_params;
                            dtp.ALLOWNOTEXIST = ALLOWNOTEXIST;
                            dtp.BATCHSIZE = BATCHSIZE;
                            dtp.THREADCOUNT = THREADCOUNT;
                            PList.Add(dtp);
                        }
                        else
                            BulkSourceThread(function_id, task_id, tradedate, exch_id, BUSI_DATE, db_source, db_target_list, exec_params, imp_file, filename_source_list, SOURCE_TABLE_NAME, TARGET_TABLE_NAME, sql_exp2, OP_EXP, op_exp_list, TRAN_MODE, map_params, ALLOWNOTEXIST, BATCHSIZE, THREADCOUNT, false);
                    }
                }
                if (PList.Count > 0)//多线程
                {
                    if (db_source_list.Count > 1 && PList.Count > 2 && PList.Count > db_source_list.Count)
                        PList = SortRandom(PList);
                    ParallelOptions opt = new ParallelOptions();
                    opt.MaxDegreeOfParallelism = PList[0].THREADCOUNT;
                    Parallel.ForEach(PList, opt, (item, state) =>//并行
                    {
                        if (taskException != null)
                            state.Break();
                        else
                        {
                            if (bwt != null && bwt.CancellationPending)
                                taskException = new Exception("线程被取消");
                            else
                            {
                                item.THREADCOUNT = item.THREADCOUNT / PList.Count;
                                BulkSourceThread(function_id, task_id, tradedate, exch_id, item.BUSI_DATE, item.DB_SOURCE, db_target_list, exec_params, item.IMP_FILE, item.FILENAME_SOURCE_LIST, item.SOURCE_TABLE_NAME, item.TARGET_TABLE_NAME, item.SQL_EXP, item.OP_EXP, item.OP_EXP_LIST, item.TRAN_MODE, item.MAP_PARAMS, item.ALLOWNOTEXIST, item.BATCHSIZE, item.THREADCOUNT, true);
                            }
                        }
                    });
                }
                if (IsFileLog) WriteLog(exch_id, "结束执行任务", true);
                if (taskException != null)
                    throw new Exception(taskException.Message, taskException);
                if (ido > 0)
                {
                    uint add_recordcount = 0;
                    FileCombineInfo fci = null;
                    foreach (string filename in exp_filelist.Keys)//文件合并
                    {
                        if (bwt != null && bwt.CancellationPending)
                            throw new Exception("线程被取消");
                        fci = exp_filelist[filename];
                        using (FileStream fileOut = new FileStream(fci.FileName, FileMode.Append))
                        {
                            using (FileStream fileIn = new FileStream(filename, FileMode.Open, FileAccess.Read))//读取临时文件，写入到目标文件
                            {
                                if (fci.Position > 0) fileIn.Position = fci.Position;
                                int ilen = 0;
                                byte[] buff = new byte[1024];
                                while ((ilen = fileIn.Read(buff, 0, 1024)) > 0)
                                    fileOut.Write(buff, 0, ilen);
                            }
                        }
                        if (fci.RecordCount > 0) add_recordcount += fci.RecordCount;
                        File.Delete(filename);
                    }
                    if (add_recordcount > 0 && fci.FileType.Equals(DataFileType.DBF))//DBF文件
                    {
                        DbfFile odbf = null;
                        try
                        {
                            odbf = new DbfFile();
                            odbf.Open(fci.FileName, FileMode.Open);
                            odbf.Header.RecordCount += add_recordcount;//修改文件头的记录数
                        }
                        finally
                        {
                            if (odbf != null) odbf.Close();
                        }
                    }
                    if (task_id.Length > 0)
                    {
                        string msg = "";
                        if (ERROR_MSG.Length > 0) msg += ERROR_MSG;
                        else if (file_msg.Length > 0) msg += file_msg;
                        if (msg.Length == 0) msg = "已成功。";
                        if (is_change_over) msg += "(切换到备机采集)";
                        if (FILE_TIME.Length > 0) msg = FILE_TIME + msg;
                        WriteLog(zkbif, begin_time, task_id, exch_id, function_id, tradedate, msg);//写本方LOG
                    }
                }
            }
            return icount;
        }

        private void BulkSourceThread(string function_id, string task_id, string tradedate, string exch_id, string BUSI_DATE, DbItem db_source, DbItemCollection db_target_list, Dictionary<string, string> exec_params, bool imp_file, List<string> filename_source_list, string SOURCE_TABLE_NAME, string TARGET_TABLE_NAME, string SQL_EXP, string OP_EXP, List<string> op_exp_list, string TRAN_MODE, string map_params, bool ALLOWNOTEXIST, int batchsize, int THREADCOUNT, bool isthread)
        {
            try
            {
                List<string> sql_list = new List<string>();
                DataFileType dft = GetDataFileType("", SQL_EXP);
                if (dft.Equals(DataFileType.DLL))
                    sql_list.Add(SQL_EXP);
                else if (db_source.DB_TYPE.Equals(DataBaseType.MongoDB) || db_source.DB_TYPE.Equals(DataBaseType.Kafka))
                    sql_list.Add(SOURCE_TABLE_NAME);
                else
                {
                    BaseInterFace sif = new BaseInterFace(db_source);
                    string sql = "";
                    int idx1 = SOURCE_TABLE_NAME.IndexOf("%");
                    int idx2 = SOURCE_TABLE_NAME.IndexOf("[");
                    if (SOURCE_TABLE_NAME.EndsWith("%") || SOURCE_TABLE_NAME.EndsWith("]") && idx2 >= 0 && db_source.DB_TYPE.Equals(DataBaseType.SQLServer))//表名是包含通配符
                    {
                        string sql_sel = "select name from sys.tables with(nolock) where name like '" + SOURCE_TABLE_NAME + "'";
                        if (IsFileLog) WriteLog(exch_id, "数据库编号：" + db_source.DB_NO.ToString() + "，" + sql_sel, true);
                        DataTable tab_name_list = sif.ExecSelect(sql_sel);
                        if (tab_name_list.Rows.Count == 0)
                        {
                            if (ALLOWNOTEXIST)
                                return;
                            else
                                throw new Exception("没有找到任何表");
                        }
                        sql = "";
                        int idx3 = (idx1 < idx2 ? idx1 : idx2);
                        if (idx3 < 0) idx3 = (idx1 > idx2 ? idx1 : idx2);
                        for (int x = 0; x < tab_name_list.Rows.Count; x++)
                        {
                            string table_name = tab_name_list.Rows[x][0].ToString();
                            string nodeid = table_name.Substring(idx3);
                            if (sql.Length > 0) sql += " UNION ALL ";
                            if (SQL_EXP.Length > 0)
                            {
                                if (SQL_EXP.Trim().ToUpper().StartsWith("SELECT"))
                                {
                                    string tmpsql = SQL_EXP.Replace(ATTR_SOURCE_TABLE_NAME, table_name).Replace(ATTR_NODE_ID, nodeid);
                                    if (ReturnCreateTableSQL)
                                        sql += "SELECT * FROM (" + tmpsql + ") X WHERE 1=2";
                                    else
                                        sql += tmpsql;
                                }
                                else
                                {
                                    if (ReturnCreateTableSQL)
                                        sql += "SELECT * FROM " + table_name + " WHERE (" + SQL_EXP + ") AND 1=2";
                                    else
                                        sql += "SELECT * FROM " + table_name + " WHERE " + SQL_EXP;
                                }
                            }
                            else
                            {
                                sql += "SELECT * FROM " + table_name;
                                if (ReturnCreateTableSQL) sql += " WHERE 1=2";
                            }
                            if (exec_params.ContainsKey("NOT_UNION"))
                            {
                                sql_list.Add(sql);
                                sql = "";
                            }
                        }
                    }
                    else
                    {
                        if (SQL_EXP.Length > 0)
                        {
                            if (SQL_EXP.Trim().ToUpper().StartsWith("SELECT") || SQL_EXP.Trim().ToUpper().StartsWith("EXEC ") || SQL_EXP.Trim().ToUpper().StartsWith("CALL ") || SQL_EXP.Trim().StartsWith("@"))
                            {
                                sql = SQL_EXP;
                                if (ReturnCreateTableSQL && SQL_EXP.Trim().ToUpper().StartsWith("SELECT"))
                                    sql = "SELECT * FROM (" + sql + ") X WHERE 1=2";
                            }
                            else
                            {
                                if (ReturnCreateTableSQL)
                                    sql = "SELECT * FROM " + ATTR_SOURCE_TABLE_NAME + " WHERE (" + SQL_EXP + ") AND 1=2";
                                else
                                    sql = "SELECT * FROM " + ATTR_SOURCE_TABLE_NAME + " WHERE " + SQL_EXP;
                            }
                        }
                        else
                        {
                            sql = "SELECT * FROM " + ATTR_SOURCE_TABLE_NAME;
                            if (ReturnCreateTableSQL) sql += " WHERE 1=2";
                        }
                        if (filename_source_list.Count == 0 && !exec_params.ContainsKey(ATTR_SOURCE_TABLE_NAME))
                            sql = sql.Replace(ATTR_SOURCE_TABLE_NAME, SOURCE_TABLE_NAME);
                    }
                    Dictionary<string, List<string>> exec_params2 = new Dictionary<string, List<string>>();
                    foreach (string param_name in exec_params.Keys)
                    {
                        if (exec_params[param_name].Contains(CommFuns.StringSplitChar.ToString()))
                            exec_params2.Add(param_name, DataFuns.SplitString(exec_params[param_name], CommFuns.StringSplitChar, false));
                        else
                            sql = sql.Replace(param_name, Funcs.ReplaceWord(exec_params[param_name], BUSI_DATE));
                    }
                    if (exec_params2.Count > 0)
                    {
                        int item_count = 0;
                        foreach (List<string> item in exec_params2.Values)
                        {
                            if (item.Count > item_count) item_count = item.Count;
                        }
                        string sql2 = "";
                        for (int i = 0; i < item_count; i++)
                        {
                            if (i > 0) sql2 += " UNION ALL ";
                            string stmp = sql;
                            foreach (string key in exec_params2.Keys)
                            {
                                stmp = stmp.Replace(key, exec_params2[key][i]);
                            }
                            if (stmp.Trim().ToUpper().StartsWith("EXEC ") || stmp.Trim().ToUpper().StartsWith("CALL "))
                                sql_list.Add(stmp);
                            else
                                sql2 += stmp;
                        }
                        sql = sql2;
                    }
                    if (sql_list.Count == 0) sql_list.Add(sql);
                }
                List<DataTaskParam> PList = new List<DataTaskParam>();
                if (IsFileLog) WriteLog(exch_id, "开始执行BulkTargetThread", true);
                lock (ForCount_Lock) { if (!ForCount.ContainsKey("3")) ForCount.Add("3", db_target_list.Count); }
                foreach (DbItem db_target in db_target_list)
                {
                    for (int j = 0; j < sql_list.Count; j++)
                    {
                        string sql2 = sql_list[j];
                        if (bwt != null && bwt.CancellationPending)
                            throw new Exception("线程被取消");
                        if (ReturnCreateTableSQL && CreateTableSQL.Length > 0)//返回SQL建表脚本只执行一次
                            break;
                        if ((db_target_list.Count > 1 || sql_list.Count > 1) && THREADCOUNT > 1)//多线程
                        {
                            DataTaskParam dtp = new DataTaskParam();
                            dtp.DB_TARGET = db_target;
                            dtp.SQL = sql2;
                            PList.Add(dtp);
                        }
                        else
                            BulkTargetThread(function_id, task_id, tradedate, exch_id, BUSI_DATE, db_source, db_target, exec_params, imp_file, filename_source_list, sql2, TARGET_TABLE_NAME, SQL_EXP, OP_EXP, op_exp_list, TRAN_MODE, map_params, ALLOWNOTEXIST, batchsize, THREADCOUNT, isthread);
                    }
                }
                if (PList.Count > 0)//多线程
                {
                    if (db_target_list.Count > 1 && PList.Count > 2 && PList.Count > db_target_list.Count)
                        PList = SortRandom(PList);
                    ParallelOptions opt = new ParallelOptions();
                    opt.MaxDegreeOfParallelism = THREADCOUNT;
                    Parallel.ForEach(PList, opt, (item, state) =>//并行
                    {
                        if (taskException != null)
                            state.Break();
                        else
                        {
                            if (bwt != null && bwt.CancellationPending)
                                taskException = new Exception("线程被取消");
                            else
                            {
                                int iTHREADCOUNT = THREADCOUNT / PList.Count;
                                BulkTargetThread(function_id, task_id, tradedate, exch_id, BUSI_DATE, db_source, item.DB_TARGET, exec_params, imp_file, filename_source_list, item.SQL, TARGET_TABLE_NAME, SQL_EXP, OP_EXP, op_exp_list, TRAN_MODE, map_params, ALLOWNOTEXIST, batchsize, iTHREADCOUNT, true);
                            }
                        }
                    });
                }
                if (IsFileLog) WriteLog(exch_id, "结束执行BulkTargetThread", true);
            }
            catch (Exception ex)
            {
                if (isthread)
                    taskException = ex;
                else
                    throw;
            }
        }

        private void BulkTargetThread(string function_id, string task_id, string tradedate, string exch_id, string BUSI_DATE, DbItem db_source, DbItem db_target, Dictionary<string, string> exec_params, bool imp_file, List<string> filename_source_list, string sql, string TARGET_TABLE_NAME, string SQL_EXP, string OP_EXP, List<string> op_exp_list, string TRAN_MODE, string map_params, bool ALLOWNOTEXIST, int batchsize, int THREADCOUNT, bool isthread)
        {
            BaseInterFace tif = null;
            string new_table_name = "";
            bool drop_flag = false;
            try
            {
                lock (ExecuteInfo_Lock)
                {
                    if (!ExecuteInfoList.ContainsKey(db_target))//没有执行状态信息
                    {
                        ExecuteInfoList.Add(db_target, new ExecuteInfo());
                        if (ExecuteTotalCount == 0)
                        {
                            ExecuteTotalCount = 1;
                            foreach (int i in ForCount.Values)//计算总共执行次数
                            {
                                ExecuteTotalCount = ExecuteTotalCount * i;
                            }
                        }
                    }
                }
                DateTime begin_time = DateTime.Now;
                bool exp_file = db_target.DB_TYPE.Equals(DataBaseType.Dbf);
                string filename_target = "";
                if (exp_file)
                {
                    filename_target = ReplaceNetPath(Funcs.ReplaceWord(Path.Combine(db_target.DB_IP, db_target.DB_NAME, TARGET_TABLE_NAME), BUSI_DATE));
                    AutoLoginNet(filename_target, db_target.USER_NAME, db_target.USER_PASSWORD);
                }
                if (imp_file && exp_file)
                {
                    bool isret = false;
                    for (int itmp = 0; itmp < filename_source_list.Count; itmp++)//多文件
                    {
                        string filename_source = filename_source_list[itmp];
                        if (filename_source.Equals(filename_target))
                            continue;
                        if (FilePath.IsURL(filename_source))
                        {
                            Dictionary<string, string> param = DataFuns.SplitStringValue(SQL_EXP, ',');
                            OpenURL.DownloadFile(filename_source, filename_target, GetProxyIp(param), GetProxyPort(param));
                            isret = true;
                            continue;
                        }
                        FileInfo sfile = new FileInfo(filename_target);
                        if (new FileInfo(filename_source).Extension.Equals(sfile.Extension) && SQL_EXP.Length == 0)//文件复制
                        {
                            if (sfile.Exists) sfile.Delete();
                            if (!sfile.Directory.Exists) sfile.Directory.Create();
                            File.Copy(filename_source, filename_target, true);
                            isret = true;
                            continue;
                        }
                    }
                    if (isret) return;
                }
                tif = new BaseInterFace(db_target);//创建目标数据库连接
                if (TIMEOUT >= 0) tif.CommandTimeout = TIMEOUT;
                int idx1 = TARGET_TABLE_NAME.IndexOf("%");
                int idx2 = TARGET_TABLE_NAME.IndexOf("[");
                Dictionary<string, string> table_list = new Dictionary<string, string>();
                if (TARGET_TABLE_NAME.EndsWith("%") || TARGET_TABLE_NAME.EndsWith("]") && idx2 >= 0 && db_target.DB_TYPE.Equals(DataBaseType.SQLServer))//表名是包含通配符
                {
                    string sql_sel = "select name from sys.tables where name like '" + TARGET_TABLE_NAME + "'";
                    if (IsFileLog) WriteLog(exch_id, "数据库编号：" + db_target.DB_NO.ToString() + "，" + sql_sel, true);
                    DataTable tab_name_list = tif.ExecSelect(sql_sel);
                    if (tab_name_list.Rows.Count == 0)
                    {
                        if (ReturnCreateTableSQL)
                            table_list.Add("#", "");
                        else if (ALLOWNOTEXIST)
                            return;
                        else
                            throw new Exception("没有找到任何表");
                    }
                    int idx3 = (idx1 < idx2 ? idx1 : idx2);
                    if (idx3 < 0) idx3 = (idx1 > idx2 ? idx1 : idx2);
                    for (int x = 0; x < tab_name_list.Rows.Count; x++)
                    {
                        string table_name = tab_name_list.Rows[x][0].ToString();
                        string nodeid = table_name.Substring(idx3);
                        table_list.Add(table_name, nodeid);
                    }
                }
                else
                    table_list.Add(TARGET_TABLE_NAME, "");

                bool exec_sql = true;
                List<string> op_exp_list2 = new List<string>();
                if (TARGET_TABLE_NAME.StartsWith("#") && TARGET_TABLE_NAME.Length > 1 && !ReturnMemDataTable)//临时表
                {
                    bool all_temp_table = true;
                    for (int i = 0; i < op_exp_list.Count; i++)
                    {
                        if (!op_exp_list[i].Contains(TARGET_TABLE_NAME) && !op_exp_list[i].Contains(ATTR_TARGET_TABLE_NAME))
                        {
                            all_temp_table = false;
                            break;
                        }
                    }
                    if (all_temp_table)//后置SQL全部涉及临时表，可以放到线程里去创建临时表
                    {
                        op_exp_list2 = op_exp_list;
                        exec_sql = false;
                    }
                    else//只创建一个临时表
                    {
                        lock (ExecuteInfo_Lock)
                        {
                            if (ExecuteInfoList[db_target].TEMP_TABLE_NAME.Length == 0)
                            {
                                new_table_name = CreateTempTable(exch_id, tif, db_target, TARGET_TABLE_NAME);
                                ExecuteInfoList[db_target].TEMP_TABLE_NAME = new_table_name;
                            }
                            else
                                new_table_name = ExecuteInfoList[db_target].TEMP_TABLE_NAME;
                        }
                    }
                }
                foreach (string key in table_list.Keys)
                {
                    if (bwt != null && bwt.CancellationPending)
                        throw new Exception("线程被取消");
                    if (ReturnCreateTableSQL && CreateTableSQL.Length > 0)
                        break;
                    string nodeid = table_list[key];
                    string sql_op = "";
                    string table_name = (new_table_name.Length > 0 ? new_table_name : key);
                    if (OP_EXP.Length > 0)
                    {
                        if (OP_EXP.EndsWith("1=2"))
                            sql_op = "";
                        else if (OP_EXP.Trim().ToUpper().StartsWith("SELECT") || OP_EXP.Trim().ToUpper().StartsWith("EXEC ") || OP_EXP.Trim().ToUpper().StartsWith("CALL "))
                            sql_op = OP_EXP;
                        else if (OP_EXP.Trim().ToUpper().StartsWith("TRUNCATE") || OP_EXP.Trim().ToUpper().StartsWith("DELETE"))
                        {
                            if (OP_EXP.ToUpper().Equals("DELETE"))
                                sql_op = "DELETE FROM " + table_name;
                            else
                                sql_op = OP_EXP.Replace(ATTR_TARGET_TABLE_NAME, table_name);
                        }
                        else
                            sql_op = "DELETE FROM " + table_name + " WHERE " + OP_EXP;
                    }
                    else if (!table_name.StartsWith("#"))
                    {
                        sql_op = "TRUNCATE TABLE " + table_name;
                        if (db_target.DB_TYPE.Equals(DataBaseType.DB2)) sql_op += " IMMEDIATE";
                    }
                    List<DataTaskParam> PList = new List<DataTaskParam>();
                    int itmp = (filename_source_list.Count == 0 ? -1 : 0);
                    for (; itmp < filename_source_list.Count; itmp++)//多文件
                    {
                        if (bwt != null && bwt.CancellationPending)
                            throw new Exception("线程被取消");
                        if (ReturnCreateTableSQL && CreateTableSQL.Length > 0)//返回SQL建表脚本只执行一次
                            break;
                        string filename_source = (itmp < 0 ? "" : filename_source_list[itmp]);
                        DbItem db_source2 = db_source;
                        string sql2 = sql;
                        if (filename_source.Length > 0 && !FilePath.IsURL(filename_source) && !FilePath.IsFTP(filename_source))
                        {
                            db_source2 = new DbItem();
                            FileInfo file = new FileInfo(filename_source);
                            db_source2.DB_TYPE = DataBaseType.Dbf;
                            db_source2.DB_IP = file.Directory.FullName;
                            if (sql2.Contains(ATTR_SOURCE_TABLE_NAME))
                                sql2 = sql2.Replace(ATTR_SOURCE_TABLE_NAME, file.Name);
                        }
                        string SQL_EXP2 = SQL_EXP;
                        foreach (string param_name in exec_params.Keys)
                        {
                            if (exec_params[param_name].Contains(CommFuns.StringSplitChar.ToString()))
                            {
                                List<string> exec_params2 = DataFuns.SplitString(exec_params[param_name], CommFuns.StringSplitChar, false);
                                int idx = (itmp < 0 || itmp >= exec_params2.Count ? 0 : itmp);
                                SQL_EXP2 = SQL_EXP2.Replace(param_name, Funcs.ReplaceWord(exec_params2[idx], BUSI_DATE));
                            }
                            else
                                SQL_EXP2 = SQL_EXP2.Replace(param_name, Funcs.ReplaceWord(exec_params[param_name], BUSI_DATE));
                        }
                        if (IsFileLog) WriteLog(exch_id, "开始执行BulkMultiThread", true);
                        if (imp_file && filename_source_list.Count > 1 && THREADCOUNT > 1)//多文件同时读
                        {
                            DataTaskParam dtp = new DataTaskParam();
                            dtp.DB_SOURCE = db_source2;
                            dtp.FILENAME_SOURCE = filename_source;
                            dtp.SQL = sql2;
                            dtp.SQL_EXP = SQL_EXP2;
                            PList.Add(dtp);
                        }
                        else
                            BulkMultiThread(function_id, task_id, tradedate, exch_id, db_source2, db_target, imp_file, exp_file, filename_source, filename_target, table_name, nodeid, sql2, sql_op, SQL_EXP2, OP_EXP, op_exp_list2, TRAN_MODE, map_params, batchsize, THREADCOUNT, isthread);
                    }
                    if (PList.Count > 0)//多线程
                    {
                        ParallelOptions opt = new ParallelOptions();
                        opt.MaxDegreeOfParallelism = THREADCOUNT;
                        Parallel.ForEach(PList, opt, (item, state) =>//并行
                        {
                            if (taskException != null)
                                state.Break();
                            else
                            {
                                if (bwt != null && bwt.CancellationPending)
                                    taskException = new Exception("线程被取消");
                                else
                                {
                                    int iTHREADCOUNT = THREADCOUNT / PList.Count;
                                    BulkMultiThread(function_id, task_id, tradedate, exch_id, item.DB_SOURCE, db_target, imp_file, exp_file, item.FILENAME_SOURCE, filename_target, table_name, nodeid, item.SQL, sql_op, item.SQL_EXP, OP_EXP, op_exp_list2, TRAN_MODE, map_params, batchsize, iTHREADCOUNT, true);
                                }
                            }
                        });
                    }
                    if (IsFileLog) WriteLog(exch_id, "结束执行BulkMultiThread", true);
                }
                bool islast = false;//本数据库最后一次执行
                lock (ExecuteInfo_Lock)
                {
                    ExecuteInfoList[db_target].EXECUTE_COUNT++;
                    if (ExecuteInfoList[db_target].EXECUTE_COUNT == ExecuteTotalCount) islast = true;
                }
                if (!ReturnMemDataTable && islast)
                {
                    if (exec_sql && !exp_file && taskException == null)//最后执行的线程，并且不是导出文件时，才要处理后置SQL
                    {
                        string tmp_table_name = (new_table_name.Length > 0 ? new_table_name : TARGET_TABLE_NAME);
                        ExecuteAfterSQL(exch_id, function_id, tradedate, tif, db_target, tmp_table_name, TARGET_TABLE_NAME, op_exp_list);
                    }
                    drop_flag = true;
                    DropTempTable(exch_id, tif, db_target, new_table_name);
                    if (WRITE_LOG && task_id.Length > 0)//写对方LOG
                    {
                        try
                        {
                            string msg = "已成功。";
                            if (ERROR_MSG.Length > 0) msg = ERROR_MSG;
                            WriteLog(tif, begin_time, task_id, exch_id, function_id, tradedate, msg);
                        }
                        catch (Exception ex2)
                        {
                            Funcs.WriteErrorLog(ex2);
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                if (!drop_flag) DropTempTable(exch_id, tif, db_target, new_table_name);
                if (isthread)
                    taskException = ex;
                else
                    throw;
            }
        }

        /// <summary>
        /// 单文件或单表的多线程
        /// </summary>
        private void BulkMultiThread(string function_id, string task_id, string tradedate, string exch_id, DbItem db_source, DbItem db_target, bool imp_file, bool exp_file, string filename_source, string filename_target, string table_name, string nodeid, string sql, string sql_op, string SQL_EXP, string OP_EXP, List<string> op_exp_list, string TRAN_MODE, string map_params, int batchsize, int THREADCOUNT, bool isthread)
        {
            try
            {
                if (IsFileLog) WriteLog(exch_id, "开始执行ExectueInsertThread", true);
                if (THREADCOUNT > 1)//多线程
                {
                    List<string> sqllist = null;
                    if (!imp_file && sql.Trim().ToUpper().StartsWith("SELECT"))
                    {
                        if (SPLIT_COLNAME.Length > 0)//数据源是数据库、并指定分片字段
                        {
                            sqllist = new List<string>();
                            DataTable dt = null;
                            bool is_max_min_split = false;
                            BaseInterFace bif = new BaseInterFace(db_source);
                            if (SPLIT_COLNAME.StartsWith(":"))//采用最大最小值平均切片
                            {
                                is_max_min_split = true;
                                dt = new DataTable();
                                SPLIT_COLNAME = SPLIT_COLNAME.Substring(1);
                                string split_sql = "SELECT MAX(CASE WHEN " + SPLIT_COLNAME + " IS NULL THEN 1 ELSE 0 END) AS NULL_VALUE,MIN(" + SPLIT_COLNAME + ") AS MIN_VALUE,MAX(" + SPLIT_COLNAME + ") AS MAX_VALUE FROM (" + sql + ") X";//求最大最小值
                                DataTable dtval = bif.ExecSelect(split_sql);
                                if (dtval.Rows.Count > 0)
                                {
                                    DataRow dr = dtval.Rows[0];
                                    long MAX_VALUE = 0;
                                    if (long.TryParse(dr["MAX_VALUE"].ToString(), out MAX_VALUE))
                                    {
                                        long MIN_VALUE = 0;
                                        long.TryParse(dr["MIN_VALUE"].ToString(), out MIN_VALUE);
                                        long avg_rows = (MAX_VALUE - MIN_VALUE) / THREADCOUNT;
                                        avg_rows++;
                                        dt.Columns.Add("COLNAME", dtval.Columns["MAX_VALUE"].DataType);
                                        dt.Columns.Add("CNT", dtval.Columns["MAX_VALUE"].DataType);
                                        if (dtval.Rows[0]["NULL_VALUE"].ToString().Equals("1"))
                                            dt.Rows.Add(new object[] { DBNull.Value, 1 });
                                        for (int i = 1; i <= THREADCOUNT; i++)
                                        {
                                            dt.Rows.Add(new object[] { MIN_VALUE + avg_rows * i, avg_rows });
                                        }
                                    }
                                }
                            }
                            else//分组计算记录数，并排序
                            {
                                string split_sql = "SELECT " + SPLIT_COLNAME + " AS COLNAME,COUNT(*) AS CNT FROM (" + sql + ") X GROUP BY " + SPLIT_COLNAME + " ORDER BY COLNAME";
                                dt = bif.ExecSelect(split_sql);
                            }
                            if (dt.Rows.Count > 1)//至少一组以上
                            {
                                string chr = dt.Columns["COLNAME"].DataType.IsValueType ? "" : "'";//字符类型的字段要加单引号
                                string begin = "SELECT * FROM (" + sql + ") X WHERE " + SPLIT_COLNAME;
                                List<string> splitlist = new List<string>();
                                List<int> splitcount = new List<int>();
                                long total_rows = long.Parse(dt.Compute("SUM(CNT)", "true").ToString());//计算总记录数
                                long avg_rows = total_rows / THREADCOUNT;//每个分片的平均记录数
                                long add_rows = 0;
                                int icount = 0;
                                bool has_null = false;//分组值里有NULL
                                DataRow[] drs = dt.Select("COLNAME IS NULL");
                                if (drs.Length > 0)
                                {
                                    has_null = true;
                                    DataRow dr = drs[0];
                                    int idx = dt.Rows.IndexOf(dr);
                                    if (idx > 0)
                                    {
                                        DataRow drnew = dt.NewRow();
                                        drnew.ItemArray = dr.ItemArray;
                                        dt.Rows.Remove(dr);
                                        dt.Rows.InsertAt(drnew, 0);
                                    }
                                }
                                for (int i = 0; i < dt.Rows.Count; i++)
                                {
                                    icount++;
                                    add_rows += long.Parse(dt.Rows[i]["CNT"].ToString());//当前分片已累计记录数
                                    if (dt.Rows.Count - splitlist.Count <= THREADCOUNT || add_rows >= avg_rows || i == dt.Rows.Count - 1)//分组小于线程数、累计达到每片平均记录数、最后一个组
                                    {
                                        splitlist.Add(dt.Rows[i]["COLNAME"].ToString());
                                        splitcount.Add(is_max_min_split ? 2 : icount);
                                        total_rows -= add_rows;
                                        add_rows = 0;
                                        icount = 0;
                                        if (THREADCOUNT - splitlist.Count > 0)
                                            avg_rows = total_rows / (THREADCOUNT - splitlist.Count);//重新计算剩余记录数的每个分片的平均记录数
                                    }
                                }
                                for (int i = 0; i < splitlist.Count; i++)
                                {
                                    if (splitcount[i] == 1)//该分片只有一个值
                                    {
                                        if (i == 0 && has_null)
                                            sqllist.Add(begin + " IS NULL");
                                        else
                                            sqllist.Add(begin + "=" + chr + splitlist[i].Replace("'", "''") + chr);
                                    }
                                    else//自动分片
                                    {
                                        if (i == 0)
                                            sqllist.Add(begin + "<=" + chr + splitlist[i].Replace("'", "''") + chr + (has_null ? " OR " + SPLIT_COLNAME + " IS NULL" : ""));
                                        else
                                            sqllist.Add(begin + ">" + chr + splitlist[i - 1].Replace("'", "''") + chr + " AND " + SPLIT_COLNAME + "<=" + chr + splitlist[i].Replace("'", "''") + chr);
                                        //if (i == splitlist.Count - 1)
                                        //    sqllist.Add(begin + ">" + chr + splitlist[i].Replace("'", "''") + chr);
                                    }
                                }
                            }
                            else
                                THREADCOUNT = 1;
                        }
                        else
                        {
                            ERROR_MSG += (ERROR_MSG.Length > 0 ? "，" : "提示信息：") + "THREADCOUNT大于1，但未定义SPLIT_COLNAME参数";
                            THREADCOUNT = 1;//改为单线程
                        }
                    }
                    if (THREADCOUNT > 1)
                    {
                        ParallelOptions opt = new ParallelOptions();
                        opt.MaxDegreeOfParallelism = THREADCOUNT;//设置线程数量，具体执行由操作系统控制（CPU核数）
                        if (sqllist == null)//数据源是文件类型（例如BCP、DBF），没有分片，自动采用记录行号求余数的读法
                        {
                            Parallel.For(0, THREADCOUNT, opt, (REMAINDER, state) =>//启用多线程
                            {
                                if (taskException != null)
                                    state.Break();
                                else
                                    ExectueInsertThread(function_id, task_id, tradedate, exch_id, db_source, db_target, imp_file, exp_file, filename_source, filename_target, table_name, nodeid, sql, sql_op, SQL_EXP, OP_EXP, op_exp_list, TRAN_MODE, map_params, batchsize, REMAINDER, THREADCOUNT, true);
                            });
                        }
                        else if (sqllist.Count > 0)//数据源是数据库，有分片
                        {
                            Parallel.ForEach(sqllist, opt, (sql2, state) =>//启用多线程
                            {
                                if (taskException != null)
                                    state.Break();
                                else
                                    ExectueInsertThread(function_id, task_id, tradedate, exch_id, db_source, db_target, imp_file, exp_file, filename_source, filename_target, table_name, nodeid, sql2, sql_op, SQL_EXP, OP_EXP, op_exp_list, TRAN_MODE, map_params, batchsize, -1, 1, true);
                            });
                        }
                    }
                }
                if (THREADCOUNT <= 1)
                    ExectueInsertThread(function_id, task_id, tradedate, exch_id, db_source, db_target, imp_file, exp_file, filename_source, filename_target, table_name, nodeid, sql, sql_op, SQL_EXP, OP_EXP, op_exp_list, TRAN_MODE, map_params, batchsize, -1, 1, isthread);
                if (IsFileLog) WriteLog(exch_id, "结束执行ExectueInsertThread", true);
            }
            catch (Exception ex)
            {
                if (isthread)
                    taskException = ex;
                else
                    throw;
            }
        }
        /// <summary>
        /// 插入数据线程
        /// </summary>
        private void ExectueInsertThread(string function_id, string task_id, string tradedate, string exch_id, DbItem db_source, DbItem db_target, bool imp_file, bool exp_file, string filename_source, string filename_target, string table_name, string nodeid, string sql, string sql_op, string SQL_EXP, string OP_EXP, List<string> op_exp_list, string TRAN_MODE, string map_params, int batchsize, int REMAINDER, int THREADCOUNT, bool isthread)
        {
            IDataReader dtread = null;
            string tmp_filename_source = "";
            bool is_tmp_file = false;
            string new_table_name = "";
            BaseInterFace tif = null;
            DataFileType dft = GetDataFileType(filename_source, SQL_EXP);
            try
            {
                bool do_remainder = false;
                Dictionary<string, string> param = new Dictionary<string, string>();
                if (imp_file || db_source.DB_TYPE.Equals(DataBaseType.MongoDB) || db_source.DB_TYPE.Equals(DataBaseType.Kafka))
                {
                    param = DataFuns.SplitStringValue(SQL_EXP.Replace(ATTR_SOURCE_TABLE_NAME, filename_source).Replace(ATTR_TARGET_TABLE_NAME, filename_target), ',');
                    if (REMAINDER >= 0 && (dft.Equals(DataFileType.BCP) || dft.Equals(DataFileType.CSV) || dft.Equals(DataFileType.DBF)))
                    {
                        param.Add("REMAINDER", REMAINDER.ToString());//多线程的余数
                        param["THREADCOUNT"] = THREADCOUNT.ToString();
                        do_remainder = true;
                    }
                }
                if (FilePath.IsURL(filename_source) || FilePath.IsFTP(filename_source))
                {
                    if (param.ContainsKey("FILE_NAME") || dft.Equals(DataFileType.BCP) || dft.Equals(DataFileType.CSV) || dft.Equals(DataFileType.XLS) || dft.Equals(DataFileType.DBF) || dft.Equals(DataFileType.DOC))
                    {
                        if (param.ContainsKey("FILE_NAME"))
                            tmp_filename_source = param["FILE_NAME"].Trim();
                        if (tmp_filename_source.Length == 0)
                        {
                            is_tmp_file = true;
                            string dirSep = DataFuns.DirSeparator;
                            tmp_filename_source = AbstractInterFace.StartupPath + "LOG" + dirSep;
                            if (task_id.Length == 0 || isthread)
                                tmp_filename_source += AbstractInterFace.GetGuidString();
                            else
                                tmp_filename_source += task_id;
                            tmp_filename_source += "." + dft.ToString().ToLower();
                        }
                        if (FilePath.IsFTP(filename_source))
                        {
                            FtpServerInfo ser = new FtpServerInfo(filename_source);
                            ser.UsePassive = true;
                            ser.DownloadFile(tmp_filename_source);
                        }
                        else
                        {
                            if (param.ContainsKey("POST"))
                            {
                                Encoding encoding = Encoding.UTF8;
                                if (param.ContainsKey("CODEPAGE") && param["CODEPAGE"].Length > 0)
                                    encoding = CommFuns.GetEncoding(param["CODEPAGE"]);
                                string contents = GetURLtoString(filename_source, param, encoding);
                                File.WriteAllText(tmp_filename_source, contents);
                            }
                            else
                                OpenURL.DownloadFile(filename_source, tmp_filename_source, GetProxyIp(param), GetProxyPort(param));//自动下载
                        }
                        filename_source = tmp_filename_source;
                    }
                }
                string execute_sql = "";
                if (dft.Equals(DataFileType.DLL))//动态调用DLL
                {
                    string forwarddll = DataFuns.SplitStringValue(SQL_EXP.Replace(ATTR_SOURCE_DB_NO, db_source.DB_NO.ToString()).Replace(ATTR_TARGET_DB_NO, db_target.DB_NO.ToString()), ',')["DLL"];
                    forwarddll = forwarddll.Replace(ATTR_SOURCE_TABLE_NAME, filename_source).Replace(ATTR_TARGET_TABLE_NAME, filename_target);
                    if (IsSqlLog)
                        WriteLog(exch_id, function_id, tradedate, forwarddll);//写LOG
                    if (IsFileLog) WriteLog(exch_id, "调用DLL：" + forwarddll, true);
                    int idx = forwarddll.IndexOf("#");
                    object[] execParams = new object[] { };
                    if (idx > 0)
                    {
                        execParams = DataFuns.SplitString(forwarddll.Substring(idx + 1), ',', true).ToArray();
                        forwarddll = forwarddll.Substring(0, idx);
                    }
                    object obj = InvokeDynamic.CreateObject(forwarddll);
                    if (IsWeb) InvokeDynamic.SetPropertyInfo(obj, "NetPathMapTable", NetPathMapTable);
                    dtread = (IDataReader)InvokeDynamic.InvokeMember(obj, execParams);
                }
                else if (dft.Equals(DataFileType.XML))//XML文件
                    dtread = GetXMLDataReader(filename_source, param);
                else if (dft.Equals(DataFileType.JSON))//JSON文件
                    dtread = GetJSONDataReader(filename_source, param);
                else if (dft.Equals(DataFileType.HTML))//HTML文件
                    dtread = GetHTMLDataReader(filename_source, param, SQL_EXP);
                else if (dft.Equals(DataFileType.PDF))//PDF文件
                    dtread = GetPDFDataReader(filename_source, param);
                else if (dft.Equals(DataFileType.BCP))//BCP文本文件
                    dtread = new BCPDataReader(filename_source, param);
                else if (dft.Equals(DataFileType.CSV))//CSV文件
                    dtread = new CSVDataReader(filename_source, param);
                else if (dft.Equals(DataFileType.XLS))//XLS文件
                    dtread = GetXLSDataReader(filename_source, param);
                else if (dft.Equals(DataFileType.DBF))//DBF文件
                    dtread = new DbfDataReader(filename_source, param);
                else if (dft.Equals(DataFileType.DOC))//DOC文件
                    dtread = GetDOCDataReader(filename_source, param);
                else if (db_source.DB_TYPE.Equals(DataBaseType.MongoDB))//MongoDB
                    dtread = new MongoDBDataReader(db_source, sql, param);
                else if (db_source.DB_TYPE.Equals(DataBaseType.Kafka))//Kafka
                    dtread = new KafkaDataReader(db_source, sql, param);
                else
                {
                    sql = sql.Replace(ATTR_NODE_ID, nodeid).Replace(ATTR_SOURCE_DB_NO, db_source.DB_NO.ToString()).Replace(ATTR_TARGET_DB_NO, db_target.DB_NO.ToString());
                    if (IsSqlLog)
                        WriteLog(exch_id, function_id, tradedate, sql);//写LOG
                    object[] PMS = CommFuns.SplitSQL(db_source.DB_TYPE, sql);//分拆SQL语句
                    if (sql.Trim().ToUpper().StartsWith("CALL ") && PMS[1] != null)//如果调用Oracle存储过程后再执行查询SQL
                    {
                        DbParamCollection pcl = (DbParamCollection)PMS[1];
                        if (pcl.Count > 0)
                        {
                            int idx = pcl.Count - 1;
                            if (pcl[idx].Value != null)
                            {
                                string val = pcl[idx].Value.ToString();
                                if (val.Trim().ToUpper().StartsWith("SELECT "))
                                {
                                    pcl.RemoveAt(idx);
                                    BaseInterFace bif = new BaseInterFace(db_source);
                                    if (TIMEOUT >= 0) bif.CommandTimeout = TIMEOUT;
                                    if (IsFileLog) WriteLog(exch_id, "数据库编号：" + db_source.DB_NO.ToString() + "," + (string)PMS[0], true);
                                    bif.ExecUpdate((string)PMS[0], pcl, (bool)PMS[2]);
                                    if (val.Trim().ToUpper().Equals("SELECT"))
                                        return;
                                    else
                                        PMS = new object[] { val, null, false };
                                }
                            }
                        }
                    }
                    ///////重做3次//////
                    int iredo = 0;
                    int iredomax = 3;
                    while ((iredo++) < iredomax)
                    {
                        try
                        {
                            BaseInterFace bif = new BaseInterFace(db_source);
                            if (TIMEOUT >= 0) bif.CommandTimeout = TIMEOUT;
                            execute_sql = (string)PMS[0];
                            bool is_pro = (bool)PMS[2];
                            if (IsFileLog) WriteLog(exch_id, "获取数据源，数据库编号：" + db_source.DB_NO.ToString() + "," + execute_sql, true);
                            dtread = bif.ExecReader(execute_sql, (DbParamCollection)PMS[1], is_pro);//创建数据流
                            if (is_pro) execute_sql = "";
                            break;
                        }
                        catch (Exception ex)
                        {
                            if (db_source.GROUP_NO < 0 && CheckShutdown(ex))//切换到备库（集群组号是负数，并且主库已挂机）
                            {
                                is_change_over = true;//切换标志
                                db_source = GetDbItem(zkbif, SOURCE_DB_LIST, (0 - db_source.GROUP_NO).ToString());//获取备库
                            }
                            else
                            {
                                bool isredo = CheckReDo(ex);
                                if (!isredo || iredo == iredomax)
                                    throw;
                                Thread.Sleep(3000);
                            }
                        }
                    }
                    ////////////////////
                }
                if (REMAINDER >= 0 && !do_remainder)
                {
                    string split = SPLIT_COLNAME;
                    if (split.Length == 0 && dtread.FieldCount > 0) split = dtread.GetName(0);
                    dtread = new FilterDataReader(dtread, REMAINDER, THREADCOUNT, split);//过滤数据流
                    do_remainder = true;
                }
                bool IsAllColumnSame = false;
                string StrColumnMap = "";
                if (ReturnCreateTableSQL)
                    CheckSchemaTable(db_source, db_target, table_name, map_params, dtread, exp_file, ref IsAllColumnSame, ref StrColumnMap);
                else if (ReturnMemDataTable)
                    LoadToDataTable(dtread);
                else if (exp_file)
                {
                    if (IsFileLog) WriteLog(exch_id, "开始导出文件", true);
                    ExportFile(db_source, db_target, filename_target, TRAN_MODE, map_params, dtread, OP_EXP, execute_sql, isthread);//导出文件
                    if (IsFileLog) WriteLog(exch_id, "结束导出文件", true);
                }
                else
                {
                    tif = new BaseInterFace(db_target);
                    if (TIMEOUT >= 0) tif.CommandTimeout = TIMEOUT;
                    string old_table_name = "";
                    if (table_name.StartsWith("#") && table_name.Length > 1)//临时表
                    {
                        string tmp_table_name = CreateTempTable(exch_id, tif, db_target, table_name);
                        sql_op = sql_op.Replace(table_name, tmp_table_name);
                        OP_EXP = OP_EXP.Replace(table_name, tmp_table_name);
                        old_table_name = table_name;
                        new_table_name = tmp_table_name;
                        table_name = tmp_table_name;
                    }
                    if (sql_op.Length > 0)
                    {
                        lock (ExecuteInfo_Lock)
                        {
                            if (!ExecuteInfoList[db_target].DELETE_SQL.Contains(sql_op))
                            {
                                DeleteData(exch_id, tif, db_target, sql_op);
                                ExecuteInfoList[db_target].DELETE_SQL.Add(sql_op);
                            }
                        }
                    }
                    if (dtread.FieldCount > 0)
                    {
                        if (IsFileLog) WriteLog(exch_id, "开始检测表结构", true);
                        List<ColumnMapInfo> ColMapList = CheckSchemaTable(db_source, db_target, table_name, map_params, dtread, exp_file, ref IsAllColumnSame, ref StrColumnMap);
                        if (IsFileLog) WriteLog(exch_id, "结束检测表结构", true);
                        if (!CheckHasMap(ColMapList)) throw new Exception("没有任何字段可以映射");
                        if (StrColumnMap.Length > 0)
                        {
                            map_params += StrColumnMap;
                            if (!param.ContainsKey("NODE_NAME"))
                            {
                                string msg = "存在部分字段找不到映射" + StrColumnMap;
                                if (!ERROR_MSG.Contains(msg)) ERROR_MSG += msg;
                                if (task_id.Length == 0)
                                    WriteLog(exch_id, function_id, tradedate, msg);//写LOG
                            }
                        }
                        if (db_source.DB_TYPE.Equals(DataBaseType.ClickHouse)) dtread.NextResult();
                        if (IsFileLog) WriteLog(exch_id, "开始复制数据", true);
                        bool is_insert = false;//内部批量INSERT的条件1：数据库类型相同 2：IP相同 3：用户相同 4：是SELECT语句
                        if (dft.Equals(DataFileType.DATABASE) && db_source.DB_TYPE.Equals(db_target.DB_TYPE) && db_source.DB_IP.Equals(db_target.DB_IP) && db_source.USER_NAME.Equals(db_target.USER_NAME) && execute_sql.Length > 0
                            && (!db_source.DB_TYPE.Equals(DataBaseType.Oracle) || db_source.DB_TYPE.Equals(DataBaseType.Oracle) && db_source.DB_NAME.Equals(db_target.DB_NAME)))
                        {
                            if (REMAINDER > 0 && dtread is FilterDataReader)
                                is_insert = true;
                            else
                                is_insert = InsertInto(db_source, db_target, table_name, dtread, ColMapList, execute_sql);//内部批量INSERT
                        }
                        if (!is_insert)//数据走交换平台
                        {
                            if (TRAN_MODE.Equals("3"))//文件
                            {
                                if (!db_target.DB_TYPE.Equals(DataBaseType.SQLServer) && !db_target.DB_TYPE.Equals(DataBaseType.Oracle) && !db_target.DB_TYPE.Equals(DataBaseType.DB2) && !db_target.DB_TYPE.Equals(DataBaseType.DaMeng) && !db_target.DB_TYPE.Equals(DataBaseType.SybaseIQ) && !db_target.DB_TYPE.Equals(DataBaseType.SybaseASE) && !db_target.DB_TYPE.Equals(DataBaseType.MySQL) && !db_target.DB_TYPE.Equals(DataBaseType.OceanBase) && !db_target.DB_TYPE.Equals(DataBaseType.GoldenDB))
                                    throw new Exception("目标数据库不支持文件导入");
                                if ((dft.Equals(DataFileType.BCP) || dft.Equals(DataFileType.CSV)) && REMAINDER < 0 && WRITE_THREADCOUNT <= 1)
                                {
                                    List<string> mapColumns = null;
                                    if (db_target.DB_TYPE.Equals(DataBaseType.MySQL) || db_target.DB_TYPE.Equals(DataBaseType.OceanBase) || db_target.DB_TYPE.Equals(DataBaseType.GoldenDB))
                                        mapColumns = GetMapColumns(ColMapList);
                                    FileImp(db_target, table_name, mapColumns, dft, filename_source, dtread, ColMapList, IsAllColumnSame, param, batchsize, 0, false, isthread);//文件入库
                                }
                                else
                                    FileExpImp(db_target, table_name, dtread, ColMapList, IsAllColumnSame, param, batchsize);//文件导出再入库
                            }
                            else if (TRAN_MODE.Equals("2"))//INSERT语句
                                Insert(tif, db_target, table_name, dtread, ColMapList, batchsize);//INSERT语句
                            else if (TRAN_MODE.Equals("1"))//内存表
                                CopyFromMemory(tif, db_target, table_name, dtread, ColMapList, batchsize);//从内存表批量复制
                            else//数据流
                            {
                                if (db_target.DB_TYPE.Equals(DataBaseType.MySQL) || db_target.DB_TYPE.Equals(DataBaseType.OceanBase) || db_target.DB_TYPE.Equals(DataBaseType.GoldenDB) || db_target.DB_TYPE.Equals(DataBaseType.ClickHouse))
                                {
                                    Insert(tif, db_target, table_name, dtread, ColMapList, batchsize);//批量INSERT
                                }
                                else
                                {
                                    if ((db_target.DB_TYPE.Equals(DataBaseType.Oracle) || db_target.DB_TYPE.Equals(DataBaseType.SybaseIQ) || db_target.DB_TYPE.Equals(DataBaseType.PostgreSQL) || db_target.DB_TYPE.Equals(DataBaseType.GaussDB)) && (dtread is BCPDataReader || dtread is CSVDataReader))
                                        InvokeDynamic.SetPropertyInfo(dtread, "FieldType", GetFieldType(ColMapList));//修改数据流的字段类型
                                    if (db_target.DB_TYPE.Equals(DataBaseType.PostgreSQL) || db_target.DB_TYPE.Equals(DataBaseType.GaussDB))
                                        map_params = GetMapParams(ColMapList, false);
                                    else
                                        map_params = new ColumnMap(map_params).GetMapMode(dtread);
                                    CopyFromDataReader(tif, db_target, table_name, map_params, dtread, batchsize);//从数据流批量复制
                                }
                            }
                        }
                        if (IsFileLog) WriteLog(exch_id, "结束复制数据", true);
                    }
                    if (!exp_file && taskException == null)
                    {
                        lock (AfterSQL_Lock)
                        {
                            ExecuteAfterSQL(exch_id, function_id, tradedate, tif, db_target, table_name, old_table_name, op_exp_list);//执行后置SQL
                        }
                    }
                }
                if (param.ContainsKey("ERROR_LINE"))
                    ERROR_MSG += (ERROR_MSG.Length > 0 ? "，" : "错误信息：") + "第" + param["ERROR_LINE"] + "行数据有错误";
            }
            catch (Exception ex)
            {
                if (isthread)
                    taskException = ex;
                else
                    throw;
            }
            finally
            {
                if (dtread != null && !dtread.IsClosed) dtread.Close();
                if (is_tmp_file)
                    File.Delete(tmp_filename_source);
                DropTempTable(exch_id, tif, db_target, new_table_name);
            }
        }
        /// <summary>
        /// 清理数据
        /// </summary>
        private void DeleteData(string exch_id, BaseInterFace tif, DbItem db_target, string sql_op)
        {
            object[] PMS = CommFuns.SplitSQL(db_target.DB_TYPE, sql_op);//分拆SQL语句
            string sqltmp = (string)PMS[0];
            if (IsFileLog) WriteLog(exch_id, "执行删除数据SQL，数据库编号：" + db_target.DB_NO.ToString() + "，" + sqltmp, true);
            ///////重做3次//////
            int iredo = 0;
            int iredomax = 3;
            while ((iredo++) < iredomax)
            {
                try
                {
                    tif.ExecUpdate(sqltmp, (DbParamCollection)PMS[1], (bool)PMS[2]);//清理数据
                    break;
                }
                catch (Exception ex)
                {
                    bool isredo = CheckReDo(ex);
                    if (!isredo || iredo == iredomax)
                        throw;
                    Thread.Sleep(3000);
                }
            }
            ////////////////////
        }
        /// <summary>
        /// 后置SQL处理
        /// </summary>
        private void ExecuteAfterSQL(string exch_id, string function_id, string tradedate, BaseInterFace tif, DbItem db_target, string table_name, string TARGET_TABLE_NAME, List<string> op_exp_list)
        {
            try
            {
                if (op_exp_list.Count == 0) return;
                tif.BeginTransaction();
                for (int i = 0; i < op_exp_list.Count; i++)//后置SQL的处理
                {
                    if (bwt != null && bwt.CancellationPending)
                        throw new Exception("线程被取消");
                    if (op_exp_list[i].Length > 0)
                    {
                        object[] PMS = CommFuns.SplitSQL(db_target.DB_TYPE, op_exp_list[i]);//分拆SQL语句
                        string sqltmp = ((string)PMS[0]).Replace(ATTR_TARGET_TABLE_NAME, table_name);
                        if (TARGET_TABLE_NAME.Length > 0 && !TARGET_TABLE_NAME.Equals(table_name)) sqltmp = sqltmp.Replace(TARGET_TABLE_NAME, table_name);
                        lock (sql_Lock)
                        {
                            if (IsSqlLog) WriteLog(exch_id, function_id, tradedate, sqltmp);//写LOG
                            if (IsFileLog) WriteLog(exch_id, "执行后置SQL，数据库编号：" + db_target.DB_NO.ToString() + "，" + sqltmp, true);
                            tif.ExecUpdate(sqltmp, (DbParamCollection)PMS[1], (bool)PMS[2]);
                        }
                    }
                }
                tif.Commit();
            }
            catch
            {
                tif.Rollback();
                throw;
            }
        }

        /// <summary>
        /// 创建临时表
        /// </summary>
        private string CreateTempTable(string exch_id, BaseInterFace tif, DbItem db_target, string table_name)
        {
            string tmp_table_name = AbstractInterFace.GetGuidString();
            string sqltmp = "";
            if (db_target.DB_TYPE.Equals(DataBaseType.SQLServer))
            {
                tmp_table_name = table_name.Substring(1) + tmp_table_name;
                sqltmp = "SELECT * INTO " + tmp_table_name + " FROM " + table_name.Substring(1) + " WHERE 1=2";
            }
            else
            {
                if (tmp_table_name.Length > 29) tmp_table_name = tmp_table_name.Substring(0, 29);
                tmp_table_name = 'T' + tmp_table_name;
                sqltmp = "CREATE TABLE " + tmp_table_name + " AS SELECT * FROM " + table_name.Substring(1) + " WHERE 1=2";
            }
            if (IsFileLog) WriteLog(exch_id, "创建临时表，数据库编号：" + db_target.DB_NO.ToString() + "," + sqltmp, true);
            tif.ExecUpdate(sqltmp);
            return tmp_table_name;
        }

        /// <summary>
        /// 删除临时表
        /// </summary>
        private void DropTempTable(string exch_id, BaseInterFace tif, DbItem db_target, string table_name)
        {
            if (table_name.Length > 0 && tif != null)
            {
                try
                {
                    string sqltmp = "DROP TABLE " + table_name;
                    if (IsFileLog) WriteLog(exch_id, "删除临时表，数据库编号：" + db_target.DB_NO.ToString() + "，" + sqltmp, true);
                    tif.ExecUpdate(sqltmp);//删除临时表
                }
                catch (Exception ex)
                {
                    Funcs.WriteErrorLog(ex);
                }
            }
        }
        private bool CheckShutdown(Exception ex)
        {
            //报错信息是以下几种情况就可以重做任务
            bool isbak = ex.Message.StartsWith("无法打开登录所请求的数据库")
                      || ex.Message.StartsWith("ORA-12170: TNS:Connect timeout occurred");
            return isbak;
        }

        private bool CheckReDo(Exception ex)
        {
            //报错信息是以下几种情况就可以重做任务
            //Oracle:通信通道的文件结尾，未连接到 ORACLE，连接失去联系
            //SQLServer:等待时发生超时。
            //OceanBase:You have an error in your SQL syntax
            bool isredo = (ex.Message.StartsWith("等待用于页 (") && ex.Message.EndsWith("时发生超时。"))
                        || (ex.Message.StartsWith("事务(进程") && ex.Message.EndsWith("请重新运行该事务。"))
                        || (ex.Message.StartsWith("执行超时已过期。完成操作之前已超时或服务器未响应。"))
                        || ex.Message.StartsWith("ORA-03113")
                        || ex.Message.StartsWith("ORA-03114")
                        || ex.Message.StartsWith("ORA-03135")
                        || ex.Message.StartsWith("Communication error")
                        || ex.Message.StartsWith("You have an error in your SQL syntax; check the manual that corresponds to your OceanBase");
            return isredo;
        }

        private void WriteLog(BaseInterFace bif, DateTime begin_time, string task_id, string exch_id, string function_id, string tradedate, string msg)
        {
            if (msg.Length > 4000) msg = msg.Substring(0, 4000);
            DbParamCollection sql_params = new DbParamCollection();
            sql_params.Add(new DbParam("TASK_ID", task_id, DbParamType.VarChar, false));
            sql_params.Add(new DbParam("SQLCMD_ID", exch_id, DbParamType.Int, false));
            sql_params.Add(new DbParam("FUNCTION_ID", function_id, DbParamType.Int, false));
            sql_params.Add(new DbParam("TRADEDATE", tradedate, DbParamType.VarChar, false));
            sql_params.Add(new DbParam("EXECUTE_START_TIME", begin_time, DbParamType.DateTime, false));
            sql_params.Add(new DbParam("EXECUTE_END_TIME", DateTime.Now, DbParamType.DateTime, false));
            sql_params.Add(new DbParam("LOG_MESSAGE", msg, DbParamType.NVarChar, false));
            bif.BeginTransaction();
            try
            {
                string sql = "DELETE FROM TB_SQLCMD_LOG WHERE TASK_ID=" + task_id + " AND SQLCMD_ID=" + exch_id;
                bif.ExecUpdate(sql);
                sql = "INSERT INTO TB_SQLCMD_LOG(TASK_ID,SQLCMD_ID,FUNCTION_ID,TRADEDATE,EXECUTED_FLAG,EXECUTE_START_TIME,EXECUTE_END_TIME,LOG_MESSAGE) VALUES(" + bif.ParamChar + "TASK_ID," + bif.ParamChar + "SQLCMD_ID," + bif.ParamChar + "FUNCTION_ID," + bif.ParamChar + "TRADEDATE,'2'," + bif.ParamChar + "EXECUTE_START_TIME," + bif.ParamChar + "EXECUTE_END_TIME," + bif.ParamChar + "LOG_MESSAGE)";
                bif.ExecUpdate(sql, sql_params);
                bif.Commit();
            }
            catch
            {
                bif.Rollback();
                throw;
            }
        }

        private void WriteLog(string exch_id, string function_id, string tradedate, string msg)
        {
            BaseInterFace bif = new BaseInterFace();
            sno++;
            long task_id = long.Parse(function_id + exch_id) * 1000 + sno;
            WriteLog(bif, DateTime.Now, task_id.ToString(), exch_id, function_id, tradedate, msg);
        }
        private static object privateObjectLock = new object();
        public static void WriteLog(string exch_id, string log, bool isline)
        {
            lock (privateObjectLock)
            {
                log = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + "," + "线程号:" + Thread.CurrentThread.ManagedThreadId.ToString() + "," + log;
                if (isline) log += DataFuns.NewLineString;
                string dirSep = DataFuns.DirSeparator;
                string filename = AbstractInterFace.StartupPath + "LOG" + dirSep + exch_id + ".log";
                System.IO.File.AppendAllText(filename, log);
            }
        }

        private string GetURLtoString(string url, Dictionary<string, string> param, Encoding encoding)
        {
            if (param.ContainsKey("CODEPAGE") && param["CODEPAGE"].Length > 0)
                encoding = CommFuns.GetEncoding(param["CODEPAGE"]);
            string host = GetProxyIp(param);
            int port = GetProxyPort(param);
            string post = "";
            Dictionary<string, string> header = null;
            if (param.ContainsKey("POST") && param["POST"].Length > 0)
                post = param["POST"];
            if (param.ContainsKey("HEADER") && param["HEADER"].Length > 0)
                header = DataFuns.SplitStringValue(param["HEADER"], ',');
            return OpenURL.DownloadString(url, post, header, encoding, host, port);
        }

        private string GetProxyIp(Dictionary<string, string> param)
        {
            string host = "";
            if (param.ContainsKey("PROXY_IP")) host = param["PROXY_IP"];
            return host;
        }

        private int GetProxyPort(Dictionary<string, string> param)
        {
            int port = 0;
            if (param.ContainsKey("PROXY_PORT")) int.TryParse(param["PROXY_PORT"], out port);
            return port;
        }

        private DataTableReader GetXMLDataReader(string filename_source, Dictionary<string, string> param)
        {
            DataTable srcdt = null;
            Encoding encoding = Encoding.UTF8;
            if (param.ContainsKey("CODEPAGE") && param["CODEPAGE"].Length > 0)
                encoding = CommFuns.GetEncoding(param["CODEPAGE"]);
            if (param.ContainsKey("NODE_NAME"))
            {
                string node_name = param["NODE_NAME"];
                string name_type = "0";
                if (param.ContainsKey("NAME_TYPE")) name_type = param["NAME_TYPE"];
                string xml = "";
                if (FilePath.IsURL(filename_source))
                    xml = GetURLtoString(filename_source, param, encoding);
                else
                    xml = File.ReadAllText(filename_source, encoding);
                XmlDataTable xmldt = new XmlDataTable(xml, node_name, name_type);
                srcdt = xmldt.GetDataTable();
            }
            else
            {
                DataSet dset = new DataSet();
                if (FilePath.IsURL(filename_source))
                {
                    string xml = GetURLtoString(filename_source, param, encoding);
                    if (!xml.StartsWith("<?xml") && !xml.StartsWith("<NewDataSet>"))
                        throw new Exception(xml);
                    using (StringReader sr = new StringReader(xml))
                    {
                        dset.ReadXml(sr);
                    }
                }
                else
                    dset.ReadXml(filename_source);
                if (dset.Tables.Count == 0) throw new Exception("没有找到任何表");
                string tablename = "";
                if (param.ContainsKey("TABLE_NAME")) tablename = param["TABLE_NAME"];
                srcdt = tablename.Length > 0 ? dset.Tables[tablename] : dset.Tables[0];
                if (param.ContainsKey("COL_NAME"))//扩充字段
                {
                    string col_name = param["COL_NAME"];
                    if (srcdt.Columns.Contains(col_name))
                    {
                        foreach (DataRow dr in srcdt.Rows)
                        {
                            string sep_char = srcdt.Columns[col_name].DataType.IsValueType ? "" : "'";
                            string exp = col_name + "=" + sep_char + dr[col_name].ToString() + sep_char;
                            foreach (DataTable dt in dset.Tables)
                            {
                                if (!dt.Equals(srcdt) && dt.Columns.Contains(col_name))
                                {
                                    DataRow[] drs = dt.Select(exp);
                                    if (drs.Length > 0)
                                    {
                                        foreach (DataColumn dcol in dt.Columns)
                                        {
                                            if (dcol.ColumnName.Equals(col_name)) continue;
                                            string new_col_name = (dcol.ColumnName.StartsWith(dt.TableName) ? "" : dt.TableName + "_") + dcol.ColumnName;
                                            if (!srcdt.Columns.Contains(new_col_name))
                                                srcdt.Columns.Add(new_col_name);
                                            dr[new_col_name] = drs[0][dcol];
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
            foreach (DataColumn dc in srcdt.Columns)
            {
                foreach (DataRow drd in srcdt.Rows)
                {
                    if (drd[dc.ColumnName].ToString().Length == 0)
                        drd[dc.ColumnName] = DBNull.Value;
                }
            }
            AddTableColumns(srcdt, param, filename_source);
            return srcdt.CreateDataReader();
        }

        private DataTableReader GetJSONDataReader(string filename_source, Dictionary<string, string> param)
        {
            bool ALLOWNOTEXIST = false;
            if (param.ContainsKey("ALLOWNOTEXIST") && param["ALLOWNOTEXIST"].Length > 0)
                ALLOWNOTEXIST = param["ALLOWNOTEXIST"].Equals("1");
            string TABLE_NAME = "";
            if (param.ContainsKey("TABLE_NAME"))
                TABLE_NAME = param["TABLE_NAME"].Trim();
            string FIELD_NAME = "";
            if (param.ContainsKey("FIELD_NAME"))
                FIELD_NAME = param["FIELD_NAME"].Trim();
            Encoding encoding = DataFuns.DefaultEncoding;
            if (param.ContainsKey("CODEPAGE") && param["CODEPAGE"].Length > 0)
                encoding = CommFuns.GetEncoding(param["CODEPAGE"]);
            string CONTENT = "";
            if (FilePath.IsURL(filename_source))
                CONTENT = GetURLtoString(filename_source, param, encoding);
            else
                CONTENT = File.ReadAllText(filename_source, encoding);
            JsonDataTable jdt = new JsonDataTable(CONTENT, TABLE_NAME, FIELD_NAME);
            DataTable srcdt = jdt.GetDataTable();
            if (srcdt.Rows.Count == 0)
            {
                ERROR_MSG = "找不到匹配的内容";
                if (!ALLOWNOTEXIST)
                    throw new Exception(ERROR_MSG);
            }
            AddTableColumns(srcdt, param, filename_source);
            return srcdt.CreateDataReader();
        }

        private DataTableReader GetHTMLDataReader(string filename_source, Dictionary<string, string> param, string SQL_EXP)
        {
            string PATTERN = "";//正则表达式
            if (param.ContainsKey("PATTERN"))
                PATTERN = param["PATTERN"];
            string BEGIN_PATTERN = "";
            if (param.ContainsKey("BEGIN_PATTERN") && param["BEGIN_PATTERN"].Length > 0)
                BEGIN_PATTERN = param["BEGIN_PATTERN"];
            string END_PATTERN = "";
            if (param.ContainsKey("END_PATTERN") && param["END_PATTERN"].Length > 0)
                END_PATTERN = param["END_PATTERN"];
            string REPLACE = "";
            if (param.ContainsKey("REPLACE") && param["REPLACE"].Length > 0)
                REPLACE = param["REPLACE"];
            bool ALLOWNOTEXIST = false;
            if (param.ContainsKey("ALLOWNOTEXIST") && param["ALLOWNOTEXIST"].Length > 0)
                ALLOWNOTEXIST = param["ALLOWNOTEXIST"].Equals("1");
            Encoding encoding = DataFuns.DefaultEncoding;
            if (filename_source.ToLower().EndsWith(".htm") || filename_source.ToLower().EndsWith(".html"))
                encoding = Encoding.UTF8;
            if (param.ContainsKey("CODEPAGE") && param["CODEPAGE"].Length > 0)
                encoding = CommFuns.GetEncoding(param["CODEPAGE"]);
            string contents = "";
            if (FilePath.IsURL(filename_source))
                contents = GetURLtoString(filename_source, param, encoding);
            else
                contents = File.ReadAllText(filename_source, encoding);
            if (REPLACE.Equals("1"))
            {
                contents = System.Text.RegularExpressions.Regex.Replace(contents, @"\s", "").Replace("&nbsp;", "");
                PATTERN = System.Text.RegularExpressions.Regex.Replace(PATTERN, @"\s", "").Replace("&nbsp;", "");
                BEGIN_PATTERN = System.Text.RegularExpressions.Regex.Replace(BEGIN_PATTERN, @"\s", "").Replace("&nbsp;", "");
                END_PATTERN = System.Text.RegularExpressions.Regex.Replace(END_PATTERN, @"\s", "").Replace("&nbsp;", "");
            }
            else if (REPLACE.Equals("2"))
            {
                contents = contents.Replace("\r\n", "\n");
                PATTERN = PATTERN.Replace("\r\n", "\n");
                BEGIN_PATTERN = BEGIN_PATTERN.Replace("\r\n", "\n");
                END_PATTERN = END_PATTERN.Replace("\r\n", "\n");
            }
            else if (PATTERN.Contains("\r\n") && !contents.Contains("\r\n") && contents.Contains("\n"))
                PATTERN = PATTERN.Replace("\r\n", "\n");
            DataTable srcdt = GetRegularExpDataTable(PATTERN, BEGIN_PATTERN, END_PATTERN, contents);
            if (srcdt.Rows.Count == 0)
            {
                ERROR_MSG = "找不到匹配的内容";
                if (!ALLOWNOTEXIST)
                    throw new Exception(ERROR_MSG);
            }
            AddTableColumns(srcdt, param, filename_source);
            return srcdt.CreateDataReader();
        }

        private DataTable GetRegularExpDataTable(string PATTERN, string BEGIN_PATTERN, string END_PATTERN, string contents)
        {
            DataTable srcdt = new DataTable();
            if (PATTERN.Length == 0)
            {
                srcdt.Columns.Add("CONTENT", typeof(string));
                DataRow dr = srcdt.NewRow();
                dr["CONTENT"] = contents;
                srcdt.Rows.Add(dr);
            }
            else
            {
                if (BEGIN_PATTERN.Length > 0)
                {
                    int ibegin = contents.IndexOf(BEGIN_PATTERN);
                    if (ibegin < 0)
                        contents = "";
                    else
                        contents = contents.Substring(ibegin);
                }
                if (END_PATTERN.Length > 0)
                {
                    int iend = contents.IndexOf(END_PATTERN);
                    if (iend >= 0)
                        contents = contents.Substring(0, iend);
                }
                System.Text.RegularExpressions.Regex regex = new System.Text.RegularExpressions.Regex(PATTERN, System.Text.RegularExpressions.RegexOptions.Multiline);
                string[] fields = regex.GetGroupNames();
                int itmp = 0;
                foreach (string field_name in fields)
                {
                    if (!int.TryParse(field_name, out itmp))
                        srcdt.Columns.Add(field_name);
                }
                System.Text.RegularExpressions.Match match = regex.Match(contents);
                while (match.Success)
                {
                    DataRow dr = srcdt.NewRow();
                    foreach (DataColumn dc in srcdt.Columns)
                    {
                        System.Text.RegularExpressions.Group gp = match.Groups[dc.ColumnName];
                        if (gp.Value.Length > 0) dr[dc.ColumnName] = gp.Value;
                    }
                    srcdt.Rows.Add(dr);
                    match = match.NextMatch();
                }
            }
            return srcdt;
        }

        private DataTableReader GetPDFDataReader(string filename_source, Dictionary<string, string> param)
        {
            bool NUMPAGE = false;//是否分页
            if (param.ContainsKey("NUMPAGE")) NUMPAGE = param["NUMPAGE"].Equals("1");
            string PATTERN = "";//正则表达式
            if (param.ContainsKey("PATTERN")) PATTERN = param["PATTERN"];
            string BEGIN_PATTERN = "";
            if (param.ContainsKey("BEGIN_PATTERN") && param["BEGIN_PATTERN"].Length > 0)
                BEGIN_PATTERN = param["BEGIN_PATTERN"];
            string END_PATTERN = "";
            if (param.ContainsKey("END_PATTERN") && param["END_PATTERN"].Length > 0)
                END_PATTERN = param["END_PATTERN"];
            string REPLACE = "";
            if (param.ContainsKey("REPLACE") && param["REPLACE"].Length > 0)
                REPLACE = param["REPLACE"];
            DataTable srcdt = new DataTable();
            iTextSharp.text.pdf.PdfReader pdfReader = null;
            if (FilePath.IsURL(filename_source))
            {
                using (Stream sr = OpenURL.DownloadStream(filename_source, GetProxyIp(param), GetProxyPort(param)))
                {
                    pdfReader = new iTextSharp.text.pdf.PdfReader(sr);
                }
            }
            else
                pdfReader = new iTextSharp.text.pdf.PdfReader(filename_source);
            try
            {
                int numberOfPages = pdfReader.NumberOfPages;
                if (NUMPAGE && PATTERN.Length == 0)//分页
                {
                    srcdt.Columns.Add("CONTENT", typeof(string));
                    srcdt.Columns.Add("NUMPAGE", typeof(int));
                    for (int i = 1; i <= numberOfPages; i++)
                    {
                        DataRow dr = srcdt.NewRow();
                        dr["NUMPAGE"] = i;
                        dr["CONTENT"] = iTextSharp.text.pdf.parser.PdfTextExtractor.GetTextFromPage(pdfReader, i);
                        srcdt.Rows.Add(dr);
                    }
                }
                else
                {
                    StringBuilder sbuf = new StringBuilder();
                    for (int i = 1; i <= numberOfPages; i++)
                    {
                        sbuf.Append(iTextSharp.text.pdf.parser.PdfTextExtractor.GetTextFromPage(pdfReader, i));
                        sbuf.Append("\n");
                    }
                    string contents = sbuf.ToString();
                    if (REPLACE.Equals("1"))
                    {
                        contents = System.Text.RegularExpressions.Regex.Replace(contents, @"\s", "");
                        PATTERN = System.Text.RegularExpressions.Regex.Replace(PATTERN, @"\s", "");
                        BEGIN_PATTERN = System.Text.RegularExpressions.Regex.Replace(BEGIN_PATTERN, @"\s", "");
                        END_PATTERN = System.Text.RegularExpressions.Regex.Replace(END_PATTERN, @"\s", "");
                    }
                    else if (REPLACE.Equals("2"))
                    {
                        contents = contents.Replace("\r\n", "\n");
                        PATTERN = PATTERN.Replace("\r\n", "\n");
                        BEGIN_PATTERN = BEGIN_PATTERN.Replace("\r\n", "\n");
                        END_PATTERN = END_PATTERN.Replace("\r\n", "\n");
                    }
                    else if (PATTERN.Contains("\r\n") && !contents.Contains("\r\n") && contents.Contains("\n"))
                        PATTERN = PATTERN.Replace("\r\n", "\n");
                    srcdt = GetRegularExpDataTable(PATTERN, BEGIN_PATTERN, END_PATTERN, contents);
                }
            }
            finally
            {
                pdfReader.Close();
            }
            AddTableColumns(srcdt, param, filename_source);
            return srcdt.CreateDataReader();
        }
        private DataTableReader GetXLSDataReader(string filename_source, Dictionary<string, string> param)
        {
            string read_type = "0";
            if (param.ContainsKey("READ_TYPE")) read_type = param["READ_TYPE"];
            string sheet_name = "";
            if (param.ContainsKey("SHEET_NAME")) sheet_name = param["SHEET_NAME"];
            string read_params = "";
            if (param.ContainsKey("READ_PARAMS")) read_params = param["READ_PARAMS"];
            DataTable srcdt = new Glpt.Office.ImportExcel().Read(filename_source, read_type, sheet_name, read_params);
            AddTableColumns(srcdt, param, filename_source);
            return srcdt.CreateDataReader();
        }
        private DataTableReader GetDOCDataReader(string filename_source, Dictionary<string, string> param)
        {
            DataTable srcdt = new DataTable();
            srcdt.Columns.Add("FILE_NAME");
            srcdt.Columns.Add("FILE_SIZE", typeof(long));
            srcdt.Columns.Add("FILE_CONTENT", typeof(byte[]));
            FileInfo fi = new FileInfo(filename_source);
            srcdt.Rows.Add(new object[] { fi.Name, fi.Length, File.ReadAllBytes(filename_source) });
            AddTableColumns(srcdt, param, filename_source);
            return srcdt.CreateDataReader();
        }

        /// <summary>
        /// 加载到内存表
        /// </summary>
        private void LoadToDataTable(IDataReader dtread)
        {
            lock (task_Lock)
            {
                if (MemDataTable == null)
                {
                    MemDataTable = new DataTable("SearchResult");//创建内存表
                    for (int icol = 0; icol < dtread.FieldCount; icol++)
                    {
                        DataColumn dc = new DataColumn();
                        dc.DataType = dtread.GetFieldType(icol);
                        dc.ColumnName = dtread.GetName(icol);
                        MemDataTable.Columns.Add(dc);
                    }
                }
            }
            while (dtread.Read())
            {
                if (bwt != null && bwt.CancellationPending)
                    throw new Exception("线程被取消");
                DataRow drdata = MemDataTable.NewRow();
                for (int icol = 0; icol < dtread.FieldCount; icol++)
                {
                    drdata[icol] = dtread.GetValue(icol);
                }
                lock (MemDataTable.Rows.SyncRoot)
                {
                    MemDataTable.Rows.Add(drdata);
                }
            }
        }

        /// <summary>
        /// 从内存表批量复制
        /// </summary>
        private void CopyFromMemory(BaseInterFace tif, DbItem db_target, string TARGET_TABLE_NAME, IDataReader dtread, List<ColumnMapInfo> ColMapList, int batchsize)
        {
            if (batchsize < 1)
            {
                if (db_target.DB_TYPE.Equals(DataBaseType.SybaseASE))
                    batchsize = 200;
                else if (db_target.DB_TYPE.Equals(DataBaseType.SybaseIQ))
                    batchsize = 1000;
                else
                    batchsize = 10240;
            }
            if (WRITE_THREADCOUNT < 1) WRITE_THREADCOUNT = 1;
            DataTable dtdata = new DataTable();//创建内存表
            List<int> fieldList = new List<int>();
            List<bool> valConvertList = new List<bool>();
            for (int icol = 0; icol < ColMapList.Count; icol++)
            {
                if (ColMapList[icol].ColumnOrdinal >= 0)
                {
                    DataColumn dc = new DataColumn();
                    dc.DataType = ColMapList[icol].FieldType;
                    dc.ColumnName = ColMapList[icol].ColumnName;
                    dtdata.Columns.Add(dc);
                    fieldList.Add(icol);
                    valConvertList.Add(dc.DataType.Equals(typeof(DateTime)) && !dtread.GetFieldType(icol).Equals(typeof(DateTime)));
                }
            }
            string map_mode = "0,0,";
            if (db_target.DB_TYPE.Equals(DataBaseType.PostgreSQL) || db_target.DB_TYPE.Equals(DataBaseType.GaussDB))
                map_mode = GetMapParams(ColMapList, true);
            else
                map_mode = new ColumnMap(map_mode).GetMapMode(dtdata);
            int irows = 0;
            List<System.Threading.Tasks.Task> tasklist = new List<System.Threading.Tasks.Task>();
            List<object[]> dataObj = new List<object[]>();
            while (dtread.Read())
            {
                if (bwt != null && bwt.CancellationPending)
                    throw new Exception("线程被取消");
                if (taskException != null)
                    break;
                object[] drdata = new object[dtdata.Columns.Count];
                for (int icol = 0; icol < fieldList.Count; icol++)
                {
                    drdata[icol] = dtread.GetValue(fieldList[icol]);
                    if (valConvertList[icol])
                        drdata[icol] = ConvertToDateTime(drdata[icol].ToString());
                }
                dataObj.Add(drdata);
                irows++;
                if (irows == batchsize)
                {
                    BulkCopyFromMemoryThread(WRITE_THREADCOUNT > 1 ? new BaseInterFace(db_target) : tif, TARGET_TABLE_NAME, map_mode, dtdata, dataObj, batchsize, tasklist);
                    dataObj = new List<object[]>();
                    irows = 0;
                }
            }
            if (irows > 0 && taskException == null)
                BulkCopyFromMemoryThread(WRITE_THREADCOUNT > 1 ? new BaseInterFace(db_target) : tif, TARGET_TABLE_NAME, map_mode, dtdata, dataObj, batchsize, tasklist);//最后的数据直接从内存表批量复制
            System.Threading.Tasks.Task.WaitAll(tasklist.ToArray());//等待全部线程结束
        }

        private void BulkCopyFromMemoryThread(BaseInterFace tif, string TARGET_TABLE_NAME, string map_mode, DataTable dtdata, List<object[]> dataObj, int batchsize, List<System.Threading.Tasks.Task> tasklist)
        {
            if (tasklist.Count >= WRITE_THREADCOUNT)//达到最大线程数
            {
                int idx = System.Threading.Tasks.Task.WaitAny(tasklist.ToArray());//等待线程结束
                tasklist.RemoveAt(idx);
            }
            if (tasklist.Count < WRITE_THREADCOUNT && taskException == null)
            {
                System.Threading.Tasks.Task taskitem = new System.Threading.Tasks.Task(() => BulkCopyFromMemoryThreadCopy(tif, TARGET_TABLE_NAME, map_mode, dtdata, dataObj, batchsize));
                tasklist.Add(taskitem);
                taskitem.Start();//启动
            }
        }
        private void BulkCopyFromMemoryThreadCopy(BaseInterFace tif, string TARGET_TABLE_NAME, string map_mode, DataTable dtdata, List<object[]> dataObj, int batchsize)
        {
            try
            {
                if (dataObj != null)
                {
                    DataTable table = dtdata.Clone();
                    foreach (object[] item in dataObj)
                    {
                        table.Rows.Add(item);
                    }
                    dtdata = table;
                }
                tif.BulkCopy(TARGET_TABLE_NAME, map_mode, dtdata, batchsize);
            }
            catch (Exception ex)
            {
                if (MAXERRORS > 0)
                {
                    if (dtdata.Rows.Count > 1)
                    {
                        int middle = dtdata.Rows.Count / 2;//折半递归插入
                        try
                        {
                            DataTable table = dtdata.Clone();
                            for (int i = 0; i < middle; i++)
                                table.ImportRow(dtdata.Rows[i]);
                            BulkCopyFromMemoryThreadCopy(tif, TARGET_TABLE_NAME, map_mode, table, null, batchsize);
                            table.Clear();
                            for (int i = middle; i < dtdata.Rows.Count; i++)
                                table.ImportRow(dtdata.Rows[i]);
                            BulkCopyFromMemoryThreadCopy(tif, TARGET_TABLE_NAME, map_mode, table, null, batchsize);
                        }
                        catch (Exception ex2)
                        {
                            if (ERROR_MSG.Length < 4000) ERROR_MSG += (ERROR_MSG.Length > 0 ? "，" : "错误信息：") + ex2.Message;
                        }
                    }
                    else
                    {
                        if (ERROR_MSG.Length < 4000) ERROR_MSG += (ERROR_MSG.Length > 0 ? "，" : "错误信息：") + ex.Message;
                    }
                }
                else
                    taskException = ex;
            }
        }

        /// <summary>
        /// 从数据流批量复制
        /// </summary>
        private void CopyFromDataReader(BaseInterFace tif, DbItem db_target, string TARGET_TABLE_NAME, string map_mode, IDataReader dtread, int batchsize)
        {
            if (batchsize < 1)
            {
                if (db_target.DB_TYPE.Equals(DataBaseType.SybaseASE))
                    batchsize = 200;
                else if (db_target.DB_TYPE.Equals(DataBaseType.SybaseIQ))
                    batchsize = 1000;
                else
                    batchsize = 102400;
            }
            if (WRITE_THREADCOUNT <= 1)
                tif.BulkCopy(TARGET_TABLE_NAME, map_mode, dtread, batchsize);//批量复制
            else//多线程写
            {
                ParallelOptions opt = new ParallelOptions();
                opt.MaxDegreeOfParallelism = WRITE_THREADCOUNT;
                Parallel.For(0, WRITE_THREADCOUNT, opt, (i, state) =>//并行
                {
                    if (taskException != null)
                        state.Break();
                    else
                    {
                        try
                        {
                            BaseInterFace bif = new BaseInterFace(db_target);
                            MultDataReader read = new MultDataReader(dtread);//创建一个有线程锁的数据流
                            bif.BulkCopy(TARGET_TABLE_NAME, map_mode, read, batchsize);//批量复制
                        }
                        catch (Exception ex)
                        {
                            taskException = ex;
                        }
                    }
                });
            }
        }

        /// <summary>
        /// 文件导出后再入库
        /// </summary>
        private void FileExpImp(DbItem db_target, string TARGET_TABLE_NAME, IDataReader dtread, List<ColumnMapInfo> ColMapList, bool IsAllColumnSame, Dictionary<string, string> param, int batchsize)
        {
            if (batchsize < 1) batchsize = (db_target.DB_TYPE.Equals(DataBaseType.DB2) ? 1000000 : 102400);
            if (WRITE_THREADCOUNT < 1) WRITE_THREADCOUNT = 1;
            if (dtread != null && dtread.FieldCount == 0 || ColMapList.Count == 0) return;
            List<string> mapColumns = null;
            bool is_csv = false;
            if (db_target.DB_TYPE.Equals(DataBaseType.MySQL) || db_target.DB_TYPE.Equals(DataBaseType.OceanBase) || db_target.DB_TYPE.Equals(DataBaseType.GoldenDB))
            {
                mapColumns = GetMapColumns(ColMapList);
                is_csv = true;
            }
            DataFileType dft = (is_csv ? DataFileType.CSV : DataFileType.BCP);
            string FIELDTERMINATOR = (is_csv ? "," : "\t");
            if (param.ContainsKey("FIELDTERMINATOR") && param["FIELDTERMINATOR"].Length > 0)
                FIELDTERMINATOR = DataFuns.ConvertFromHexString(param["FIELDTERMINATOR"]);
            string ROWTERMINATOR = "\r\n";
            if (param.ContainsKey("ROWTERMINATOR") && param["ROWTERMINATOR"].Length > 0)
                ROWTERMINATOR = DataFuns.ConvertFromHexString(param["ROWTERMINATOR"]).Replace("\\n", "\n").Replace("\\r", "\r");
            Encoding encoding = (is_csv ? Encoding.UTF8 : DataFuns.DefaultEncoding);
            if (param.ContainsKey("CODEPAGE") && param["CODEPAGE"].Length > 0)
                encoding = CommFuns.GetEncoding(param["CODEPAGE"]);
            string dirSep = DataFuns.DirSeparator;
            string filename = AbstractInterFace.StartupPath + "LOG" + dirSep + AbstractInterFace.GetGuidString() + (is_csv ? ".csv" : ".bcp");
            FileInfo sfile = new FileInfo(filename);
            if (sfile.Exists) sfile.Delete();
            if (!sfile.Directory.Exists) sfile.Directory.Create();
            List<string> filelist = new List<string>();
            filelist.Add(filename);
            int irows = 0;
            List<System.Threading.Tasks.Task> tasklist = new List<System.Threading.Tasks.Task>();
            StreamWriter file = new StreamWriter(filename, true, encoding);
            try
            {
                while (dtread.Read())
                {
                    if (bwt != null && bwt.CancellationPending)
                        throw new Exception("线程被取消");
                    if (taskException != null)
                        break;
                    for (int i = 0; i < dtread.FieldCount; i++)
                    {
                        if (i > 0)
                            file.Write(FIELDTERMINATOR);
                        object obj = dtread.GetValue(i);
                        if (obj != null && !obj.Equals(DBNull.Value))
                        {
                            string stmp = obj.ToString().Replace(ROWTERMINATOR, " ");
                            if (is_csv)
                            {
                                if (stmp.Contains(FIELDTERMINATOR) || stmp.Contains("\""))
                                    stmp = "\"" + stmp.Replace("\"", "\"\"") + "\"";
                            }
                            else
                                stmp = stmp.Replace(FIELDTERMINATOR, " ");

                            if (stmp.Length > 0)
                            {
                                if (obj is DateTime)
                                    stmp = ((DateTime)obj).ToString("yyyy-MM-dd HH:mm:ss.fff");
                                file.Write(stmp);
                            }
                            else
                                file.Write("\0");
                        }
                    }
                    file.Write(ROWTERMINATOR);
                    irows++;
                    if (irows == batchsize)
                    {
                        file.Close();
                        FileExpImpThread(db_target, TARGET_TABLE_NAME, mapColumns, dft, filename, dtread, ColMapList, IsAllColumnSame, param, batchsize, tasklist, batchsize);
                        filename = AbstractInterFace.StartupPath + "LOG" + dirSep + AbstractInterFace.GetGuidString() + (is_csv ? ".csv" : ".bcp");
                        filelist.Add(filename);
                        file = new StreamWriter(filename, true, encoding);
                        irows = 0;
                    }
                }
                if (irows > 0 && taskException == null)
                {
                    file.Close();
                    FileExpImpThread(db_target, TARGET_TABLE_NAME, mapColumns, dft, filename, dtread, ColMapList, IsAllColumnSame, param, batchsize, tasklist, irows);//最后的数据直接从内存表批量复制
                }
            }
            finally
            {
                file.Close();
            }
            System.Threading.Tasks.Task.WaitAll(tasklist.ToArray());//等待全部线程结束
            foreach (string fname in filelist)
            {
                if (File.Exists(fname)) File.Delete(fname);
            }
        }

        private void FileExpImpThread(DbItem db_target, string TARGET_TABLE_NAME, List<string> mapColumns, DataFileType dft, string filename, IDataReader dtread, List<ColumnMapInfo> ColMapList, bool IsAllColumnSame, Dictionary<string, string> param, int batchsize, List<System.Threading.Tasks.Task> tasklist, int data_rows)
        {
            if (tasklist.Count >= WRITE_THREADCOUNT)//达到最大线程数
            {
                int idx = System.Threading.Tasks.Task.WaitAny(tasklist.ToArray());//等待线程结束
                tasklist.RemoveAt(idx);
            }
            if (tasklist.Count < WRITE_THREADCOUNT && taskException == null)
            {
                System.Threading.Tasks.Task taskitem = new System.Threading.Tasks.Task(() => FileImp(db_target, TARGET_TABLE_NAME, mapColumns, dft, filename, dtread, ColMapList, IsAllColumnSame, param, batchsize, data_rows, true, true));
                tasklist.Add(taskitem);
                taskitem.Start();//启动
            }
        }
        private void FileImp(DbItem db_target, string TARGET_TABLE_NAME, List<string> mapColumns, DataFileType dft, string filename, IDataReader dtread, List<ColumnMapInfo> ColMapList, bool IsAllColumnSame, Dictionary<string, string> param, int batchsize, int data_rows, bool is_delete, bool isthread)
        {
            try
            {
                if (batchsize < 1) batchsize = 102400;
                string dirSep = DataFuns.DirSeparator;
                string stmpfile = AbstractInterFace.StartupPath + "LOG" + dirSep + AbstractInterFace.GetGuidString();
                bool is_csv = dft.Equals(DataFileType.CSV);
                bool has_title = (is_csv && !is_delete);//CSV带标题
                string FIELDTERMINATOR = (is_csv ? "," : "\t");
                if (param.ContainsKey("FIELDTERMINATOR") && param["FIELDTERMINATOR"].Length > 0)
                    FIELDTERMINATOR = param["FIELDTERMINATOR"];
                string ROWTERMINATOR = "\r\n";
                if (param.ContainsKey("ROWTERMINATOR") && param["ROWTERMINATOR"].Length > 0)
                    ROWTERMINATOR = param["ROWTERMINATOR"].Replace("\\n", "\n").Replace("\\r", "\r");
                Encoding encoding = (is_csv ? Encoding.UTF8 : DataFuns.DefaultEncoding);
                if (param.ContainsKey("CODEPAGE") && param["CODEPAGE"].Length > 0)
                    encoding = CommFuns.GetEncoding(param["CODEPAGE"]);
                int FIRSTROW = (has_title ? 2 : 1);
                if (param.ContainsKey("FIRSTROW") && param["FIRSTROW"].Length > 0)
                {
                    int.TryParse(param["FIRSTROW"], out FIRSTROW);
                    if (FIRSTROW < 1) FIRSTROW = 1;
                }
                int LASTROW = 0;
                if (param.ContainsKey("LASTROW") && param["LASTROW"].Length > 0)
                {
                    int.TryParse(param["LASTROW"], out LASTROW);
                }
                if (param.ContainsKey("TITLEROW") && param["TITLEROW"].Length > 0)
                {
                    int ititleIndex = 0;
                    int.TryParse(param["TITLEROW"], out ititleIndex);
                    if (ititleIndex == 0 && is_csv && FIRSTROW == 2)
                        FIRSTROW = 1;
                    else if (ititleIndex > 0 && FIRSTROW <= 1)
                        FIRSTROW = ititleIndex + 1;
                }
                if (FIRSTROW < 2) has_title = false;
                int ROWCOUNT = 0;
                if (param.ContainsKey("ROWCOUNT") && param["ROWCOUNT"].Length > 0)
                    int.TryParse(param["ROWCOUNT"], out ROWCOUNT);
                if (ROWCOUNT > 0)
                {
                    using (StreamReader stream = new StreamReader(filename, encoding))
                    {
                        int irowIndex = 0;
                        while (irowIndex < ROWCOUNT && !stream.EndOfStream)//跳到标题行
                        {
                            irowIndex++;
                            if (irowIndex == ROWCOUNT)
                            {
                                int icount = int.Parse(stream.ReadLine());
                                if (icount <= 0) return;
                                LASTROW = FIRSTROW + icount - 1;
                            }
                            else
                                stream.ReadLine();
                        }
                    }
                }
                if (db_target.DB_TYPE.Equals(DataBaseType.SQLServer))
                {
                    string control_file = "";
                    string file_arg = "";
                    if (!IsAllColumnSame)
                    {
                        FIELDTERMINATOR = DataFuns.ConvertFromHexString(FIELDTERMINATOR);
                        ROWTERMINATOR = DataFuns.ConvertFromHexString(ROWTERMINATOR);
                        StringBuilder sb = new StringBuilder();
                        sb.AppendLine("12.0");
                        sb.AppendLine(ColMapList.Count.ToString());
                        for (int i = 0; i < ColMapList.Count; i++)
                        {
                            ColumnMapInfo dr = ColMapList[i];
                            sb.Append((dr.ColumnId + 1).ToString());
                            sb.Append("       SQLCHAR             0       ");
                            sb.Append(dr.ColumnSize.ToString());
                            sb.Append("      \"");
                            if (i < ColMapList.Count - 1)
                                sb.Append(FIELDTERMINATOR);
                            else
                                sb.Append(ROWTERMINATOR.Replace("\r", "\\r").Replace("\n", "\\n"));
                            sb.Append("\"     ");
                            sb.Append((dr.ColumnOrdinal + 1).ToString());
                            sb.Append("     ");
                            sb.Append(dr.ColumnName);
                            sb.AppendLine("                 \"\"");
                        }
                        control_file = stmpfile + ".fmt";
                        FileInfo sfile = new FileInfo(control_file);
                        if (sfile.Exists) sfile.Delete();
                        if (!sfile.Directory.Exists) sfile.Directory.Create();
                        File.WriteAllText(control_file, sb.ToString(), DataFuns.DefaultEncoding);
                        file_arg = " -f \"" + control_file + "\"";
                    }
                    else
                    {
                        file_arg = " -c";
                        if (FIELDTERMINATOR.Length > 0 && !FIELDTERMINATOR.Equals("\t")) file_arg += " -t\"" + FIELDTERMINATOR + "\"";
                        if (ROWTERMINATOR.Length > 0 && !ROWTERMINATOR.Equals("\r\n")) file_arg += " -r\"" + ROWTERMINATOR + "\"";
                    }
                    string arguments = TARGET_TABLE_NAME + " in \"" + filename + "\" -S\"" + db_target.DB_IP + "\" -U\"" + db_target.USER_NAME + "\" -P\"" + AbstractInterFace.ReplaceString(db_target.USER_PASSWORD, AbstractInterFace.PublicLeftChar, AbstractInterFace.PublicRightChar) + "\" -d\"" + db_target.DB_NAME + "\" -k" + file_arg;
                    if (FIRSTROW > 1) arguments += " -F " + FIRSTROW.ToString();
                    if (LASTROW > 1) arguments += " -L " + LASTROW.ToString();
                    if (MAXERRORS > 0) arguments += " -m " + MAXERRORS.ToString();
                    if (batchsize > 0) arguments += " -b " + batchsize.ToString();
                    try
                    {
                        CommFuns.Execute("bcp", arguments);
                    }
                    finally
                    {
                        try
                        {
                            if (control_file.Length > 0 && File.Exists(control_file)) File.Delete(control_file);
                        }
                        catch { }
                    }
                }
                else if (db_target.DB_TYPE.Equals(DataBaseType.SybaseASE))
                {
                    string control_file = "";
                    string file_arg = "";
                    if (!IsAllColumnSame)
                    {
                        FIELDTERMINATOR = DataFuns.ConvertFromHexString(FIELDTERMINATOR);
                        ROWTERMINATOR = DataFuns.ConvertFromHexString(ROWTERMINATOR);
                        StringBuilder sb = new StringBuilder();
                        sb.AppendLine("12.0");
                        sb.AppendLine(ColMapList.Count.ToString());
                        for (int i = 0; i < ColMapList.Count; i++)
                        {
                            ColumnMapInfo dr = ColMapList[i];
                            sb.Append((dr.ColumnId + 1).ToString());
                            sb.Append("       SQLCHAR             0       ");
                            sb.Append(dr.ColumnSize.ToString());
                            sb.Append("      \"");
                            if (i < ColMapList.Count - 1)
                                sb.Append(FIELDTERMINATOR);
                            else
                                sb.Append(ROWTERMINATOR.Replace("\r", "\\r").Replace("\n", "\\n"));
                            sb.Append("\"     ");
                            sb.Append((dr.ColumnOrdinal + 1).ToString());
                            sb.Append("     ");
                            sb.Append(dr.ColumnName);
                            sb.AppendLine("                 \"\"");
                        }
                        control_file = stmpfile + ".fmt";
                        FileInfo sfile = new FileInfo(control_file);
                        if (sfile.Exists) sfile.Delete();
                        if (!sfile.Directory.Exists) sfile.Directory.Create();
                        File.WriteAllText(control_file, sb.ToString(), DataFuns.DefaultEncoding);
                        file_arg = " -f \"" + control_file + "\"";
                    }
                    else
                    {
                        file_arg = " -c";
                        if (FIELDTERMINATOR.Length > 0 && !FIELDTERMINATOR.Equals("\t")) file_arg += " -t\"" + FIELDTERMINATOR + "\"";
                        if (ROWTERMINATOR.Length > 0 && !ROWTERMINATOR.Equals("\r\n")) file_arg += " -r\"" + ROWTERMINATOR + "\"";
                    }
                    string arguments = db_target.DB_NAME + ".dbo." + TARGET_TABLE_NAME + " in \"" + filename + "\" -S\"" + db_target.DB_IP + "\" -U\"" + db_target.USER_NAME + "\" -P\"" + AbstractInterFace.ReplaceString(db_target.USER_PASSWORD, AbstractInterFace.PublicLeftChar, AbstractInterFace.PublicRightChar) + "\" -J cp936" + file_arg;
                    if (FIRSTROW > 1) arguments += " -F " + FIRSTROW.ToString();
                    if (LASTROW > 1) arguments += " -L " + LASTROW.ToString();
                    if (MAXERRORS > 0) arguments += " -m " + MAXERRORS.ToString();
                    if (batchsize > 0) arguments += " -b " + batchsize.ToString();
                    try
                    {
                        CommFuns.Execute("bcp", arguments);
                    }
                    finally
                    {
                        try
                        {
                            if (control_file.Length > 0 && File.Exists(control_file)) File.Delete(control_file);
                        }
                        catch { }
                    }
                }
                else if (db_target.DB_TYPE.Equals(DataBaseType.Oracle))
                {
                    StringBuilder sb = new StringBuilder();
                    sb.AppendLine("load data");
                    sb.AppendLine("infile '" + filename + "'");
                    sb.AppendLine("into table " + TARGET_TABLE_NAME + " append");
                    sb.Append("fields terminated by ");
                    if (FIELDTERMINATOR.StartsWith("0x"))
                        sb.AppendLine("x'" + FIELDTERMINATOR.Replace("0x", "") + "'");
                    else
                        sb.AppendLine("'" + FIELDTERMINATOR + "'");
                    sb.AppendLine("trailing nullcols");
                    sb.AppendLine("(");
                    for (int i = 0; i < ColMapList.Count; i++)
                    {
                        ColumnMapInfo dr = ColMapList[i];
                        sb.Append(dr.ColumnName);
                        if (dr.ColumnSize > 255)
                            sb.Append(" char(" + dr.ColumnSize.ToString() + ")");
                        if (dr.ColumnOrdinal < 0)
                            sb.Append(" FILLER");
                        if (i < ColMapList.Count - 1)
                            sb.AppendLine(",");
                        else
                            sb.AppendLine(")");
                    }
                    string control_file = stmpfile + ".ctl";
                    string log_file = stmpfile + ".log";
                    string bad_file = stmpfile + ".bad";
                    FileInfo sfile = new FileInfo(control_file);
                    if (sfile.Exists) sfile.Delete();
                    if (!sfile.Directory.Exists) sfile.Directory.Create();
                    File.WriteAllText(control_file, sb.ToString(), DataFuns.DefaultEncoding);
                    string[] conn_ip_port = db_target.DB_IP.Split(new char[] { ':' }, StringSplitOptions.RemoveEmptyEntries);
                    string arguments = "userid=" + db_target.USER_NAME + "/" + AbstractInterFace.ReplaceString(db_target.USER_PASSWORD, AbstractInterFace.PublicLeftChar, AbstractInterFace.PublicRightChar) + "@//" + conn_ip_port[0] + ":" + (conn_ip_port.Length > 1 ? conn_ip_port[1] : "1521") + "/" + db_target.DB_NAME;
                    if (FIRSTROW - 1 > 0) arguments += " skip=" + (FIRSTROW - 1).ToString();
                    if (LASTROW - FIRSTROW + 1 > 0) arguments += " load=" + (LASTROW - FIRSTROW + 1).ToString();
                    try
                    {
                        ///////重做3次//////
                        int iredo = 0;
                        int iredomax = 3;
                        while ((iredo++) < iredomax)
                        {
                            try
                            {
                                CommFuns.Sqlldr(arguments, control_file, log_file, bad_file);
                                break;
                            }
                            catch (Exception ex)
                            {
                                if (!ex.Message.StartsWith("ORA-00054") || iredo == iredomax)//资源正忙或者超时失效
                                    throw;
                                Thread.Sleep(3000);
                            }
                        }
                        ////////////////////
                    }
                    finally
                    {
                        try
                        {
                            if (File.Exists(control_file)) File.Delete(control_file);
                            if (File.Exists(log_file)) File.Delete(log_file);
                            if (File.Exists(bad_file)) File.Delete(bad_file);
                        }
                        catch { }
                    }
                }
                else if (db_target.DB_TYPE.Equals(DataBaseType.DaMeng))
                {
                    StringBuilder sb = new StringBuilder();
                    sb.AppendLine("load data");
                    sb.AppendLine("infile '" + filename + "'");
                    sb.AppendLine("into table " + TARGET_TABLE_NAME);
                    sb.Append("fields ");
                    if (FIELDTERMINATOR.StartsWith("0x"))
                        sb.AppendLine("x'" + FIELDTERMINATOR.Replace("0x", "") + "'");
                    else
                        sb.AppendLine("'" + FIELDTERMINATOR + "'");
                    sb.AppendLine("(");
                    for (int i = 0; i < ColMapList.Count; i++)
                    {
                        ColumnMapInfo dr = ColMapList[i];
                        sb.Append(dr.ColumnName);
                        if (dr.ColumnSize > 255)
                            sb.Append(" char(" + dr.ColumnSize.ToString() + ")");
                        if (dr.ColumnOrdinal < 0)
                            sb.Append(" FILLER");
                        if (i < ColMapList.Count - 1)
                            sb.AppendLine(",");
                        else
                            sb.AppendLine(")");
                    }
                    string control_file = stmpfile + ".ctl";
                    string log_file = stmpfile + ".log";
                    string bad_file = stmpfile + ".bad";
                    FileInfo sfile = new FileInfo(control_file);
                    if (sfile.Exists) sfile.Delete();
                    if (!sfile.Directory.Exists) sfile.Directory.Create();
                    File.WriteAllText(control_file, sb.ToString(), DataFuns.DefaultEncoding);
                    string[] conn_ip_port = db_target.DB_IP.Split(new char[] { ':' }, StringSplitOptions.RemoveEmptyEntries);
                    string arguments = "userid=" + db_target.USER_NAME + "/" + AbstractInterFace.ReplaceString(db_target.USER_PASSWORD, AbstractInterFace.PublicLeftChar, AbstractInterFace.PublicRightChar) + "@" + conn_ip_port[0] + ":" + (conn_ip_port.Length > 1 ? conn_ip_port[1] : "5236");
                    if (FIRSTROW - 1 > 0) arguments += " skip=" + (FIRSTROW - 1).ToString();
                    if (LASTROW - FIRSTROW + 1 > 0) arguments += " load=" + (LASTROW - FIRSTROW + 1).ToString();
                    try
                    {
                        CommFuns.Dmfldr(arguments, control_file, log_file, bad_file);
                    }
                    finally
                    {
                        try
                        {
                            if (File.Exists(control_file)) File.Delete(control_file);
                            if (File.Exists(log_file)) File.Delete(log_file);
                            if (File.Exists(bad_file)) File.Delete(bad_file);
                        }
                        catch { }
                    }
                }
                else if (db_target.DB_TYPE.Equals(DataBaseType.SybaseIQ))
                {
                    StringBuilder sb = new StringBuilder();
                    sb.AppendLine("USE " + db_target.DB_NAME);
                    sb.AppendLine("go");
                    sb.AppendLine("LOAD TABLE " + TARGET_TABLE_NAME);
                    sb.AppendLine("(");
                    for (int i = 0; i < ColMapList.Count; i++)
                    {
                        ColumnMapInfo dr = ColMapList[i];
                        sb.Append(dr.ColumnName + " '");
                        if (i < ColMapList.Count - 1)
                        {
                            sb.Append(FIELDTERMINATOR.StartsWith("0x") ? FIELDTERMINATOR.Replace("0x", "\\X") : FIELDTERMINATOR);
                            sb.AppendLine("',");
                        }
                        else
                        {
                            if (ROWTERMINATOR.Length == 0)
                                sb.Append("\\X0A");
                            else
                                sb.Append(ROWTERMINATOR.StartsWith("0x") ? ROWTERMINATOR.Replace("0x", "\\X") : ROWTERMINATOR);
                            sb.AppendLine("')");
                        }
                    }
                    sb.AppendLine("USING CLIENT FILE '" + filename + "'");
                    sb.AppendLine("QUOTES OFF");
                    sb.AppendLine("ESCAPES OFF");
                    sb.AppendLine("FORMAT ascii");
                    sb.AppendLine("STRIP OFF");
                    sb.AppendLine("WITH CHECKPOINT OFF");
                    sb.AppendLine("go");
                    sb.AppendLine("COMMIT");
                    sb.AppendLine("go");
                    sb.AppendLine("exit");
                    string control_file = stmpfile + ".SQL";
                    FileInfo sfile = new FileInfo(control_file);
                    if (sfile.Exists) sfile.Delete();
                    if (!sfile.Directory.Exists) sfile.Directory.Create();
                    File.WriteAllText(control_file, sb.ToString(), DataFuns.DefaultEncoding);
                    string[] conn_ip_port = db_target.DB_IP.Split(new char[] { ':' }, StringSplitOptions.RemoveEmptyEntries);
                    string userid = "\"uid=" + db_target.USER_NAME + ";pwd=" + AbstractInterFace.ReplaceString(db_target.USER_PASSWORD, AbstractInterFace.PublicLeftChar, AbstractInterFace.PublicRightChar) + "\" -host " + conn_ip_port[0] + " -port " + (conn_ip_port.Length > 1 ? conn_ip_port[1] : "2638");
                    try
                    {
                        CommFuns.Dbisql(userid, control_file);
                    }
                    finally
                    {
                        try
                        {
                            if (File.Exists(control_file)) File.Delete(control_file);
                        }
                        catch { }
                    }
                }
                else if (db_target.DB_TYPE.Equals(DataBaseType.MySQL) || db_target.DB_TYPE.Equals(DataBaseType.OceanBase) || db_target.DB_TYPE.Equals(DataBaseType.GoldenDB))
                {
                    int irow = data_rows;
                    System.Threading.Tasks.Task<int> taskitem = null;
                    if (irow == 0)
                    {
                        taskitem = new System.Threading.Tasks.Task<int>(() => GetFileRowCount(dtread));
                        taskitem.Start();//启动
                    }
                    Dictionary<string, string> param2 = new Dictionary<string, string>();
                    foreach (string key in param.Keys)
                    {
                        if (!key.Equals("CODEPAGE") && !key.Equals("FIRSTROW"))
                            param2.Add(key, param[key]);
                    }
                    param2.Add("CODEPAGE", encoding.Equals(Encoding.UTF8) ? "utf8" : "gb2312");
                    if (!param2.ContainsKey("FIELDTERMINATOR")) param2.Add("FIELDTERMINATOR", FIELDTERMINATOR);
                    if (FIRSTROW > 1) param2.Add("FIRSTROW", (FIRSTROW - 1).ToString());
                    int loadcount = 0;
                    ///////重做3次//////
                    int iredo = 0;
                    int iredomax = 3;
                    while ((iredo++) < iredomax)
                    {
                        try
                        {
                            if (new FileInfo(filename).Length > 0)
                            {
                                if (db_target.DB_TYPE.Equals(DataBaseType.MySQL))
                                    loadcount = new MySqlInterFace(db_target).LoadFile(TARGET_TABLE_NAME, filename, mapColumns, param2);
                                else
                                    loadcount = new OceanBaseInterFace(db_target).LoadFile(TARGET_TABLE_NAME, filename, mapColumns, param2);
                            }
                            break;
                        }
                        catch (Exception ex)
                        {
                            if (!ex.Message.StartsWith("Fatal error encountered attempting to read the resultset.") || iredo == iredomax)
                            {
                                throw;
                            }
                            Thread.Sleep(3000);
                        }
                    }
                    ////////////////////
                    if (taskitem != null)
                    {
                        taskitem.Wait();
                        irow = taskitem.Result - FIRSTROW + 1;
                        if (has_title) irow++;
                        if (irow < 0) irow = 0;
                    }
                    if (irow != loadcount)
                    {
                        string msg = "总记录数" + irow.ToString() + "条,但只加载了" + loadcount.ToString() + "条";
                        if (taskitem != null)
                            ERROR_MSG += (ERROR_MSG.Length > 0 ? "，" : "提示信息：") + msg;
                        else
                            throw new Exception(msg);
                    }
                }
                else if (db_target.DB_TYPE.Equals(DataBaseType.DB2))
                {
                    StringBuilder sb = new StringBuilder();
                    sb.AppendLine("db2 connect to " + db_target.DB_NAME + " user " + db_target.USER_NAME + " using " + AbstractInterFace.ReplaceString(db_target.USER_PASSWORD, AbstractInterFace.PublicLeftChar, AbstractInterFace.PublicRightChar));
                    sb.Append("db2 load client from " + filename + " of del modified by codepage=");
                    sb.Append(encoding.Equals(Encoding.UTF8) ? "1208" : "1386");
                    sb.Append(FIELDTERMINATOR.Equals(",") ? "" : " coldel" + DataFuns.ConvertToHexString(FIELDTERMINATOR));
                    sb.Append(is_csv ? "" : " nochardel");
                    sb.AppendLine(" insert into " + TARGET_TABLE_NAME + " ALLOW READ ACCESS");
                    sb.AppendLine("exit");
                    string bat_file = stmpfile + ".bat";
                    File.WriteAllText(bat_file, sb.ToString(), DataFuns.DefaultEncoding);
                    try
                    {
                        CommFuns.Db2CMD(bat_file);
                    }
                    finally
                    {
                        try
                        {
                            if (File.Exists(bat_file)) File.Delete(bat_file);
                        }
                        catch { }
                    }
                }
            }
            catch (Exception ex)
            {
                if (MAXERRORS > 0)
                {
                    if (ERROR_MSG.Length < 4000) ERROR_MSG += (ERROR_MSG.Length > 0 ? "，" : "错误信息：") + ex.Message + "[" + filename + "]";
                }
                else if (isthread)
                    taskException = ex;
                else
                    throw;
            }
            finally
            {
                try
                {
                    if (is_delete && File.Exists(filename)) File.Delete(filename);
                }
                catch { }
            }
        }

        private int GetFileRowCount(IDataReader dtread)
        {
            int irow = 0;
            while (dtread.Read())
            {
                irow++;
            }
            return irow;
        }

        private List<string> GetMapColumns(List<ColumnMapInfo> ColMapList)
        {
            List<string> mapColumns = new List<string>();
            for (int icol = 0; icol < ColMapList.Count; icol++)
            {
                if (ColMapList[icol].ColumnOrdinal >= 0)
                {
                    string column_name = ColMapList[icol].ColumnName;
                    mapColumns.Add(column_name);
                }
                else
                    throw new Exception("没有找到第" + (icol + 1).ToString() + "列的对应列。");
            }
            return mapColumns;
        }

        private string GetMapParams(List<ColumnMapInfo> ColMapList, bool is_datatable)
        {
            string map_params = "";
            int icol = 0;
            for (int i = 0; i < ColMapList.Count; i++)
            {
                if (ColMapList[i].ColumnOrdinal >= 0)
                {
                    if (map_params.Length > 0) map_params += ",";
                    map_params += (is_datatable ? (icol++) : i).ToString() + ":" + ColMapList[i].ColumnName + ":" + ColMapList[i].FieldType.Name;
                }
            }
            return map_params;
        }

        private Type[] GetFieldType(List<ColumnMapInfo> ColMapList)
        {
            Type[] fieldType = new Type[ColMapList.Count];
            for (int icol = 0; icol < ColMapList.Count; icol++)
            {
                fieldType[icol] = ColMapList[icol].FieldType;
            }
            return fieldType;
        }

        private bool CheckHasMap(List<ColumnMapInfo> ColMapList)
        {
            bool hasmap = false;
            for (int icol = 0; icol < ColMapList.Count; icol++)
            {
                if (ColMapList[icol].ColumnOrdinal >= 0)
                {
                    hasmap = true;
                    break;
                }
            }
            return hasmap;
        }
        private string StrCreateTableSQL = "";
        public string CreateTableSQL
        {
            get
            {
                return StrCreateTableSQL;
            }
        }
        public bool ReturnCreateTableSQL
        {
            get;
            set;
        }
        private bool _ReturnMemDataTable = false;
        public bool ReturnMemDataTable
        {
            get
            {
                return _ReturnMemDataTable || ReturnCreateTableSQL;
            }
            set
            {
                _ReturnMemDataTable = value;
            }
        }

        private DataTable GetSchemaTable(DbItem dbi, IDataReader reader)
        {
            if (dbi.DB_TYPE.Equals(DataBaseType.ClickHouse))
                return DataFuns.GetSchemaTable(reader);
            else
                return reader.GetSchemaTable();
        }

        private List<ColumnMapInfo> CheckSchemaTable(DbItem db_source, DbItem db_target, string table_name, string map_mode, IDataReader reader, bool exp_file, ref bool IsAllColumnSame, ref string StrColumnMap)
        {
            StringBuilder sbuf = new StringBuilder();
            if (table_name.StartsWith("#")) table_name = table_name.Substring(1);
            DataTable TargetTableInfo = null;
            lock (schema_Lock)
            {
                if (table_name.Length > 0 && !exp_file && !TargetTableInfoList.ContainsKey(db_target))
                {
                    string sql = "SELECT * FROM " + table_name + " WHERE 1=2";
                    BaseInterFace bif = new BaseInterFace(db_target);
                    ///////重做3次//////
                    int iredo = 0;
                    int iredomax = 3;
                    while ((iredo++) < iredomax)
                    {
                        try
                        {
                            using (IDataReader dbread = bif.ExecReader(sql))
                            {
                                TargetTableInfoList.Add(db_target, GetSchemaTable(db_target, dbread));
                            }
                            break;
                        }
                        catch (Exception ex)
                        {
                            bool isredo = CheckReDo(ex);
                            if (!isredo || iredo == iredomax)
                            {
                                if (ReturnCreateTableSQL)
                                {
                                    string msg = ex.Message.Replace("\r\n", "");
                                    int idx = msg.IndexOf("Exception:");
                                    if (idx > 0) msg = msg.Substring(idx + 10).Trim();
                                    idx = msg.IndexOf("\n");
                                    if (idx > 0) msg = msg.Substring(0, idx).Trim();
                                    sbuf.AppendLine("--无法校验表结构(" + msg + ")");
                                    break;
                                }
                                else
                                    throw;
                            }
                            Thread.Sleep(3000);
                        }
                    }
                }
                if (TargetTableInfoList.ContainsKey(db_target)) TargetTableInfo = TargetTableInfoList[db_target];
                ////////////////////
                if (TargetTableInfo != null && TargetTableInfo.Rows.Count > 0)
                {
                    int imin = int.Parse(TargetTableInfo.Compute("min(ColumnOrdinal)", "true").ToString());
                    if (imin > 0)
                    {
                        foreach (DataRow dr in TargetTableInfo.Rows)
                        {
                            dr["ColumnOrdinal"] = int.Parse(dr["ColumnOrdinal"].ToString()) - imin;
                        }
                    }
                }
            }
            ColumnMap bcm = new ColumnMap(map_mode);
            if (ReturnCreateTableSQL)
            {
                DataTable dt_schema = GetSchemaTable(db_source, reader);
                sbuf.AppendLine("CREATE TABLE " + table_name);
                sbuf.AppendLine("(");
                for (int i = 0; i < dt_schema.Rows.Count; i++)
                {
                    string col_name = dt_schema.Rows[i]["ColumnName"].ToString();
                    string coltype = GetColumnType(dt_schema.Rows[i], db_source, db_target);
                    DataRow[] drs = null;
                    string remark = "";
                    if (TargetTableInfo != null)
                    {
                        if (bcm.MAP_MODE.Equals("1"))
                        {
                            int destinationColumnId = bcm.GetDestinationColumnId(i);
                            if (destinationColumnId >= 0)
                            {
                                drs = TargetTableInfo.Select("ColumnOrdinal=" + destinationColumnId);
                                if (drs.Length > 0) col_name = drs[0]["ColumnName"].ToString();
                            }
                        }
                        else
                        {
                            string destinationColumn = bcm.GetDestinationColumn(col_name);
                            if (destinationColumn.Length > 0)
                            {
                                drs = TargetTableInfo.Select("ColumnName='" + destinationColumn.Replace("'", "''") + "'");
                            }
                        }
                        if (drs == null)
                            remark = " --已删除字段";
                        else if (drs.Length > 0)
                        {
                            string coltype2 = GetColumnType(drs[0], db_target, db_target);
                            if (!coltype2.Equals(coltype))
                                remark = " --" + coltype2;
                        }
                        else
                            remark = " --缺少字段";
                    }
                    sbuf.Append(col_name);
                    sbuf.Append(" ");
                    sbuf.Append(coltype);
                    bool allownull = true;
                    if (bool.TryParse(dt_schema.Rows[i]["AllowDBNull"].ToString(), out allownull))
                    {
                        if (!allownull) sbuf.Append(" not null");
                    }
                    sbuf.Append(i == dt_schema.Rows.Count - 1 ? "" : ",");
                    if (remark.Length > 0) sbuf.Append(remark);
                    sbuf.AppendLine();
                }
                sbuf.AppendLine(");");
                StrCreateTableSQL = sbuf.ToString();
                return null;
            }
            else
            {
                List<ColumnMapInfo> cmilist = new List<ColumnMapInfo>();
                string smap = "";
                bool wqyy = (TargetTableInfo != null && reader.FieldCount == TargetTableInfo.Rows.Count);
                for (int i = 0; i < reader.FieldCount; i++)
                {
                    string col_name = reader.GetName(i);
                    DataRow[] drs = null;
                    if (TargetTableInfo != null)
                    {
                        if (bcm.MAP_MODE.Equals("1"))
                        {
                            int destinationColumnId = bcm.GetDestinationColumnId(i);
                            if (destinationColumnId >= 0)
                            {
                                drs = TargetTableInfo.Select("ColumnOrdinal=" + destinationColumnId);
                                if (drs.Length == 0) smap += "," + i.ToString() + ":-1";
                            }
                        }
                        else
                        {
                            string destinationColumn = bcm.GetDestinationColumn(col_name);
                            if (destinationColumn.Length > 0)
                            {
                                drs = TargetTableInfo.Select("ColumnName='" + destinationColumn + "'");
                                if (drs.Length == 0) smap += "," + col_name + ":";
                            }
                        }
                    }
                    int ColumnId = i;
                    int ColumnOrdinal = -1;
                    int ColumnSize = 255;
                    Type FieldType = reader.GetFieldType(i);
                    if (drs != null && drs.Length > 0)
                    {
                        ColumnOrdinal = int.Parse(drs[0]["ColumnOrdinal"].ToString());
                        col_name = drs[0]["ColumnName"].ToString();
                        FieldType = (Type)drs[0]["DataType"];
                        if (FieldType.IsValueType)
                        {
                            if (!int.TryParse(drs[0]["NumericPrecision"].ToString(), out ColumnSize))
                                ColumnSize = int.Parse(drs[0]["ColumnSize"].ToString());
                        }
                        else
                            ColumnSize = int.Parse(drs[0]["ColumnSize"].ToString());
                    }
                    ColumnMapInfo cmi = new ColumnMapInfo();
                    cmi.ColumnId = ColumnId;
                    cmi.ColumnName = col_name;
                    cmi.ColumnOrdinal = ColumnOrdinal;
                    cmi.ColumnSize = ColumnSize;
                    cmi.FieldType = FieldType;
                    cmilist.Add(cmi);
                    if (ColumnId != ColumnOrdinal) wqyy = false;
                }
                IsAllColumnSame = wqyy;
                StrColumnMap = smap;
                return cmilist;
            }
        }

        private string GetColumnType(DataRow dr, DbItem db_source, DbItem db_target)
        {
            int max_length = 0;
            int nLength = 0;
            int nDecimals = 0;
            int.TryParse(dr["ColumnSize"].ToString(), out max_length);
            if (max_length < 0) max_length = 255;
            int.TryParse(dr["NumericPrecision"].ToString(), out nLength);
            if (nLength >= 255) nLength = 0;
            int.TryParse(dr["NumericScale"].ToString(), out nDecimals);
            if (nDecimals >= 255) nDecimals = 0;
            else if (nDecimals >= nLength) nDecimals = 2;
            Type fieldType = (dr["DataType"].Equals(DBNull.Value) ? typeof(string) : (Type)dr["DataType"]);
            string dataTypeName = "";
            if (dr.Table.Columns.Contains("DataTypeName"))
                dataTypeName = dr["DataTypeName"].ToString();
            else
            {
                if (fieldType.Equals(typeof(DateTime)))
                    dataTypeName = "date";
                else if (fieldType.Equals(typeof(Int16)) || fieldType.Equals(typeof(Int32)) || fieldType.Equals(typeof(UInt16)) || fieldType.Equals(typeof(UInt32)))
                    dataTypeName = "int";
                else if (fieldType.Equals(typeof(Int64)) || fieldType.Equals(typeof(UInt64)))
                    dataTypeName = "bigint";
                else if (fieldType.IsValueType)
                {
                    if (nDecimals > 0 || fieldType.Name.Equals("Decimal") || fieldType.Name.Equals("Double"))
                        dataTypeName = "number";
                    else
                        dataTypeName = "int";
                }
                else
                    dataTypeName = "varchar";

                if (dr.Table.Columns.Contains("PROVIDERTYPE"))
                {
                    string PROVIDERTYPE = dr["PROVIDERTYPE"].ToString();
                    if (db_source.DB_TYPE.Equals(DataBaseType.Oracle))//如果是Oracle
                    {
                        if (PROVIDERTYPE.Equals("102"))
                            dataTypeName = "blob";
                        else if (PROVIDERTYPE.Equals("104"))
                            dataTypeName = "char";
                        else if (PROVIDERTYPE.Equals("105"))
                            dataTypeName = "clob";
                        else if (PROVIDERTYPE.Equals("106"))
                            dataTypeName = "date";
                        else if (PROVIDERTYPE.Equals("107"))
                            dataTypeName = "number";
                        else if (PROVIDERTYPE.Equals("108"))
                            dataTypeName = "number";
                        else if (PROVIDERTYPE.Equals("112"))
                            dataTypeName = "int";
                        else if (PROVIDERTYPE.Equals("113"))
                            dataTypeName = "number";
                        else if (PROVIDERTYPE.Equals("117"))
                            dataTypeName = "nchar";
                        else if (PROVIDERTYPE.Equals("119"))
                            dataTypeName = "nvarchar2";
                        else if (PROVIDERTYPE.Equals("123"))
                            dataTypeName = "timestamp";
                        else if (PROVIDERTYPE.Equals("126"))
                            dataTypeName = "varchar2";
                    }
                    else if (db_source.DB_TYPE.Equals(DataBaseType.MySQL))//如果是MySQL
                    {
                        if (PROVIDERTYPE.Equals("1"))
                            dataTypeName = "tinyint";
                        else if (PROVIDERTYPE.Equals("2"))
                            dataTypeName = "smallint";
                        else if (PROVIDERTYPE.Equals("3"))
                            dataTypeName = "int";
                        else if (PROVIDERTYPE.Equals("4"))
                            dataTypeName = "float";
                        else if (PROVIDERTYPE.Equals("5"))
                            dataTypeName = "double";
                        else if (PROVIDERTYPE.Equals("7"))
                            dataTypeName = "timestamp";
                        else if (PROVIDERTYPE.Equals("8"))
                            dataTypeName = "bigint";
                        else if (PROVIDERTYPE.Equals("9"))
                            dataTypeName = "mediumint";
                        else if (PROVIDERTYPE.Equals("10"))
                            dataTypeName = "date";
                        else if (PROVIDERTYPE.Equals("11"))
                            dataTypeName = "time";
                        else if (PROVIDERTYPE.Equals("12"))
                            dataTypeName = "datetime";
                        else if (PROVIDERTYPE.Equals("13"))
                            dataTypeName = "year";
                        else if (PROVIDERTYPE.Equals("16"))
                            dataTypeName = "bit";
                        else if (PROVIDERTYPE.Equals("246"))
                            dataTypeName = "decimal";
                        else if (PROVIDERTYPE.Equals("252"))
                            dataTypeName = "blob";
                        else if (PROVIDERTYPE.Equals("253"))
                            dataTypeName = "varchar";
                        else if (PROVIDERTYPE.Equals("254"))
                            dataTypeName = "char";
                        else if (PROVIDERTYPE.Equals("600"))
                            dataTypeName = "binary";
                        else if (PROVIDERTYPE.Equals("601"))
                            dataTypeName = "varbinary";
                        else if (PROVIDERTYPE.Equals("752"))
                            dataTypeName = "text";
                    }
                }
            }
            string column_type = ChangeDataType(db_source, db_target, dataTypeName);
            if (!fieldType.Equals(typeof(Int16)) && !fieldType.Equals(typeof(Int32)) && !fieldType.Equals(typeof(Int64))
                && !fieldType.Equals(typeof(UInt16)) && !fieldType.Equals(typeof(UInt32)) && !fieldType.Equals(typeof(UInt64)) && !fieldType.Equals(typeof(DateTime))
                || dataTypeName.Equals("number"))
            {
                if (fieldType.IsValueType)
                {
                    if (!dataTypeName.Equals("int") && !dataTypeName.Equals("tinyint") && !dataTypeName.Equals("double") && !dataTypeName.Equals("float") && !dataTypeName.Equals("bit")
                        && !dataTypeName.Equals("date") && !dataTypeName.Equals("time") && !dataTypeName.Equals("datetime") && !dataTypeName.Equals("timestamp"))
                        column_type += "(" + nLength.ToString() + "," + nDecimals.ToString() + ")";
                }
                else if (!dataTypeName.Equals("clob") && !dataTypeName.Equals("blob") && !dataTypeName.Equals("text"))
                    column_type += "(" + (max_length == 2147483647 ? "max" : max_length.ToString()) + ")";
            }
            return column_type;
        }
        private string ChangeDataType(DbItem db_source, DbItem db_target, string datatype_name)
        {
            string ret = datatype_name;
            if (db_source.DB_TYPE == db_target.DB_TYPE)
                ret = datatype_name;
            else if (db_target.DB_TYPE == DataBaseType.Oracle)
            {
                if (datatype_name.Equals("varchar"))
                    ret = "varchar2";
                else if (datatype_name.Equals("character varying"))
                    ret = "varchar2";
                else if (datatype_name.Equals("nchar"))
                    ret = "char";
                else if (datatype_name.Equals("nvarchar"))
                    ret = "nvarchar2";
                else if (datatype_name.Equals("bigint"))
                    ret = "number";
                else if (datatype_name.Equals("decimal"))
                    ret = "number";
                else if (datatype_name.Equals("double"))
                    ret = "number";
                else if (datatype_name.Equals("float"))
                    ret = "number";
                else if (datatype_name.Equals("tinyint"))
                    ret = "int";
                else if (datatype_name.Equals("smallint"))
                    ret = "int";
                else if (datatype_name.Equals("mediumint"))
                    ret = "int";
                else if (datatype_name.Equals("bigint"))
                    ret = "int";
                else if (datatype_name.Equals("datetime"))
                    ret = "date";
                else if (datatype_name.Equals("datetime2"))
                    ret = "date";
                else if (datatype_name.Equals("time"))
                    ret = "date";
                else if (datatype_name.Equals("binary"))
                    ret = "raw";
                else if (datatype_name.Equals("varbinary"))
                    ret = "raw";
                else if (datatype_name.Equals("year"))
                    ret = "int";
                else if (datatype_name.Equals("bit"))
                    ret = "int";
                else if (datatype_name.Equals("text"))
                    ret = "clob";
                else if (datatype_name.Equals("image"))
                    ret = "blob";
            }
            else if (db_target.DB_TYPE == DataBaseType.SQLServer)
            {
                if (datatype_name.Equals("number"))
                    ret = "decimal";
                else if (datatype_name.Equals("mediumint"))
                    ret = "int";
                else if (datatype_name.Equals("integer"))
                    ret = "int";
                else if (datatype_name.Equals("double"))
                    ret = "float";
                else if (datatype_name.Equals("timestamp"))
                    ret = "date";
                else if (datatype_name.Equals("year"))
                    ret = "int";
                else if (datatype_name.Equals("clob"))
                    ret = "text";
                else if (datatype_name.Equals("blob"))
                    ret = "image";
                else if (datatype_name.Equals("character varying"))
                    ret = "varchar";
                else if (datatype_name.Equals("varchar2"))
                    ret = "varchar";
                else if (datatype_name.Equals("nvarchar2"))
                    ret = "nvarchar";
            }
            else if (db_target.DB_TYPE == DataBaseType.SybaseIQ)
            {
                if (datatype_name.Equals("nchar"))
                    ret = "char";
                else if (datatype_name.Equals("nvarchar"))
                    ret = "varchar";
                else if (datatype_name.Equals("character varying"))
                    ret = "varchar";
            }
            else if (db_target.DB_TYPE == DataBaseType.MySQL)
            {
                if (datatype_name.Equals("nchar"))
                    ret = "char";
                else if (datatype_name.Equals("nvarchar"))
                    ret = "varchar";
                else if (datatype_name.Equals("character varying"))
                    ret = "varchar";
                else if (datatype_name.Equals("number"))
                    ret = "decimal";
            }
            return ret;
        }
        /// <summary>
        /// 执行Insert语句
        /// </summary>
        private void Insert(BaseInterFace tif, DbItem db_target, string tablename, IDataReader reader, List<ColumnMapInfo> ColMapList, int batchsize)
        {
            bool is_oracle = db_target.DB_TYPE.Equals(DataBaseType.Oracle);
            bool is_sybaseace = db_target.DB_TYPE.Equals(DataBaseType.SybaseASE);
            bool is_clickhouse = db_target.DB_TYPE.Equals(DataBaseType.ClickHouse);
            bool is_mysql = (db_target.DB_TYPE.Equals(DataBaseType.MySQL) || db_target.DB_TYPE.Equals(DataBaseType.OceanBase) || db_target.DB_TYPE.Equals(DataBaseType.GoldenDB));
            if (batchsize < 1)
            {
                if (db_target.DB_TYPE.Equals(DataBaseType.SQLServer))
                    batchsize = 50;
                else if (is_oracle || is_sybaseace)
                    batchsize = 200;
                else if (db_target.DB_TYPE.Equals(DataBaseType.GaussDB) || db_target.DB_TYPE.Equals(DataBaseType.SybaseIQ))
                    batchsize = 1000;
                else if (is_clickhouse)
                    batchsize = 102400;
                else
                    batchsize = 500;
            }
            if (WRITE_THREADCOUNT < 1)
            {
                if (db_target.DB_TYPE.Equals(DataBaseType.SQLServer) || is_oracle || db_target.DB_TYPE.Equals(DataBaseType.PostgreSQL) || db_target.DB_TYPE.Equals(DataBaseType.GaussDB))
                    WRITE_THREADCOUNT = 10;
                else if (db_target.DB_TYPE.Equals(DataBaseType.SybaseIQ) || is_sybaseace)
                    WRITE_THREADCOUNT = 1;
                else
                    WRITE_THREADCOUNT = 5;
            }
            string insertsql = "";
            List<int> fieldList = new List<int>();
            List<DbParamType> dbpList = new List<DbParamType>();
            for (int i = 0; i < ColMapList.Count; i++)
            {
                if (ColMapList[i].ColumnOrdinal >= 0)
                {
                    string column_name = ColMapList[i].ColumnName;
                    if (fieldList.Count > 0)
                    {
                        insertsql += ",";
                    }
                    insertsql += tif.ColumnLeftChar + column_name + tif.ColumnRightChar;
                    fieldList.Add(i);
                    Type column_type = ColMapList[i].FieldType;
                    bool isbyte = column_type.Equals(typeof(byte[]));
                    if (isbyte && batchsize > 1) batchsize = 1;
                    DbParamType dbp = DbParamType.VarChar;
                    if (isbyte) dbp = DbParamType.Blob;
                    else if (column_type.Equals(typeof(DateTime))) dbp = DbParamType.DateTime;
                    else if (column_type.IsValueType) dbp = DbParamType.Double;
                    dbpList.Add(dbp);
                }
            }
            string sbegin = "insert into " + tablename + "(" + insertsql + ")";
            StringBuilder sb = new StringBuilder();
            bool is_batch = (batchsize > 1 && (is_oracle || is_sybaseace));
            if (!is_batch) sbegin += " values";
            DbParamCollection sql_params = null;
            List<object[]> datas = null;
            if (is_clickhouse) sbegin += " @bulk;";
            sb.AppendLine(sbegin);
            int irow = 0;
            List<System.Threading.Tasks.Task> tasklist = new List<System.Threading.Tasks.Task>();
            while (reader.Read())
            {
                if (bwt != null && bwt.CancellationPending)
                    throw new Exception("线程被取消");
                if (taskException != null)
                    break;
                if (is_clickhouse)
                {
                    if (irow == 0)
                    {
                        sql_params = new DbParamCollection();
                        datas = new List<object[]>();
                        sql_params.Add(new DbParam("bulk", datas, DbParamType.Blob, false));
                    }
                    object[] data = new object[fieldList.Count];
                    for (int i = 0; i < fieldList.Count; i++)
                    {
                        data[i] = reader.GetValue(fieldList[i]);
                    }
                    datas.Add(data);
                }
                else
                {
                    if (irow > 0)
                    {
                        sb.AppendLine();
                        sb.Append(is_batch ? "union all " : ",");
                    }
                    sb.Append(is_batch ? "select " : "(");
                    for (int i = 0; i < fieldList.Count; i++)
                    {
                        if (i > 0) sb.Append(",");
                        object obj = reader.GetValue(fieldList[i]);
                        if (obj == null || obj.Equals(DBNull.Value))
                        {
                            sb.Append("null");
                        }
                        else if (dbpList[i].Equals(DbParamType.Blob) || dbpList[i].Equals(DbParamType.DateTime))
                        {
                            if (sql_params == null) sql_params = new DbParamCollection();
                            string param_name = "P" + irow.ToString() + "_" + i.ToString();
                            sb.Append(tif.ParamChar + (tif.ParamChar.Equals("?") ? "" : param_name));
                            sql_params.Add(new DbParam(param_name, obj, dbpList[i], false));
                        }
                        else
                        {
                            string val = obj.ToString();
                            if (dbpList[i].Equals(DbParamType.Double))
                            {
                                if (val.Length == 0)
                                    sb.Append("null");
                                else
                                    sb.Append(val);
                            }
                            else
                            {
                                if (is_mysql) val = val.Replace("\\", "\\\\");
                                sb.Append("'" + val.Replace("'", "''''").Replace("\0", "") + "'");
                            }
                        }
                    }
                    sb.Append(is_batch ? (is_oracle ? " from dual" : "") : ")");
                }
                irow++;
                if (batchsize <= 1 || irow % batchsize == 0)
                {
                    InsertThread(WRITE_THREADCOUNT > 1 ? new BaseInterFace(db_target) : tif, sb.ToString(), sql_params, tasklist);
                    sb.Clear();
                    sb.AppendLine(sbegin);
                    sql_params = null;
                    irow = 0;
                }
            }
            if (irow > 0 && taskException == null)
            {
                InsertThread(WRITE_THREADCOUNT > 1 ? new BaseInterFace(db_target) : tif, sb.ToString(), sql_params, tasklist);
            }
            System.Threading.Tasks.Task.WaitAll(tasklist.ToArray());//等待全部线程结束
        }
        private void InsertThread(BaseInterFace tif, string sql, DbParamCollection sql_params, List<System.Threading.Tasks.Task> tasklist)
        {
            if (tasklist.Count >= WRITE_THREADCOUNT)//达到最大线程数
            {
                int idx = System.Threading.Tasks.Task.WaitAny(tasklist.ToArray());//等待线程结束
                tasklist.RemoveAt(idx);
            }
            if (tasklist.Count < WRITE_THREADCOUNT && taskException == null)
            {
                System.Threading.Tasks.Task taskitem = new System.Threading.Tasks.Task(() => InsertThreadExecUpdate(tif, sql, sql_params));
                tasklist.Add(taskitem);
                taskitem.Start();//启动
            }
        }

        private void InsertThreadExecUpdate(BaseInterFace tif, string sql, DbParamCollection sql_params)
        {
            try
            {
                tif.ExecUpdate(sql, sql_params);
            }
            catch (Exception ex)
            {
                if (MAXERRORS > 0)
                {
                    string[] sqllist = sql.Split(new string[] { DataFuns.NewLineString }, StringSplitOptions.RemoveEmptyEntries);
                    if (sqllist.Length > 2)
                    {
                        string sqltmp = "";
                        DbParamCollection dbpc = null;
                        int paramscount = 0;
                        if (sql_params != null)
                        {
                            dbpc = new DbParamCollection();
                            paramscount = sql_params.Count / (sqllist.Length - 1);
                        }
                        for (int i = 1; i < sqllist.Length; i++)
                        {
                            try
                            {
                                sqltmp = sqllist[0] + DataFuns.NewLineString;
                                if (sqllist[i].StartsWith(","))
                                    sqltmp += sqllist[i].Substring(1);
                                else if (sqllist[i].StartsWith("union all "))
                                    sqltmp += sqllist[i].Substring(10);
                                else
                                    sqltmp += sqllist[i];
                                if (dbpc != null)
                                {
                                    dbpc.Clear();
                                    for (int j = 0; j < paramscount; j++)
                                    {
                                        dbpc.Add(sql_params[(i - 1) * paramscount + j]);
                                    }
                                }
                                tif.ExecUpdate(sqltmp, dbpc);
                            }
                            catch (Exception ex2)
                            {
                                if (ERROR_MSG.Length < 4000) ERROR_MSG += (ERROR_MSG.Length > 0 ? "，" : "错误信息：") + ex2.Message + "，" + sqltmp;
                            }
                        }
                    }
                    else
                    {
                        if (ERROR_MSG.Length < 4000) ERROR_MSG += (ERROR_MSG.Length > 0 ? "，" : "错误信息：") + ex.Message + "，" + sql;
                    }
                }
                else
                    taskException = ex;
            }
        }
        /// <summary>
        /// 内部批量Insert
        /// </summary>
        private bool InsertInto(DbItem db_source, DbItem db_target, string tablename, IDataReader reader, List<ColumnMapInfo> ColMapList, string execute_sql)
        {
            bool is_ok = true;
            string insertsql = "";
            BaseInterFace tif = new BaseInterFace(db_source);
            for (int i = 0; i < ColMapList.Count; i++)
            {
                if (ColMapList[i].ColumnOrdinal >= 0)
                {
                    string column_name = ColMapList[i].ColumnName;
                    if (insertsql.Length > 0)
                        insertsql += ",";
                    insertsql += tif.ColumnLeftChar + column_name + tif.ColumnRightChar;
                }
                else
                {
                    is_ok = false;
                    break;
                }
            }
            if (is_ok)
            {
                reader.Close();
                if (!db_source.DB_NAME.Equals(db_target.DB_NAME) && !tablename.Contains(".") && db_target.DB_TYPE.Equals(DataBaseType.SQLServer))
                    tablename = db_target.DB_NAME + ".dbo." + tablename;
                string sql = "insert into " + tablename + "(" + insertsql + ") " + execute_sql;
                tif.ExecUpdate(sql);
            }
            return is_ok;
        }
        /// <summary>
        /// 导出文件
        /// </summary>
        private void ExportFile(DbItem db_source, DbItem db_target, string filename, string TRAN_MODE, string map_mode, IDataReader dtread, string op_exp, string execute_sql, bool isthread)
        {
            bool is_first = false;
            lock (ExecuteInfo_Lock)
            {
                if (ExecuteInfoList[db_target].DELETE_SQL.Count == 0)
                {
                    ExecuteInfoList[db_target].DELETE_SQL.Add("");
                    is_first = true;
                }
            }
            DataFileType dft = GetDataFileType(filename, op_exp);
            if (dft.Equals(DataFileType.DATABASE)) dft = DataFileType.BCP;
            FileInfo sfile = new FileInfo(filename);
            if (sfile.Exists && is_first) sfile.Delete();
            if (!sfile.Directory.Exists) sfile.Directory.Create();
            Dictionary<string, string> param = DataFuns.SplitStringValue(op_exp, ',');
            ColumnMap bcm = new ColumnMap(map_mode);
            List<int> fieldList = new List<int>();
            bool append = (!isthread && !is_first);//是否追加
            bool is_sql_bcp = execute_sql.Length > 0 && db_source.DB_TYPE.Equals(DataBaseType.SQLServer) && dft.Equals(DataFileType.BCP) && TRAN_MODE.Equals("3");//源数据库是SQLServer,目标数据源是BCP文件,文件方式
            if ((isthread || is_sql_bcp) && !is_first)//多线程或BCP命令导出，并且不是第一个线程
            {
                FileCombineInfo fci = new FileCombineInfo();
                fci.FileType = dft;
                fci.FileName = filename;
                filename = Path.Combine(sfile.Directory.FullName, AbstractInterFace.GetGuidString() + sfile.Extension);
                lock (exp_filelist_Lock)
                {
                    exp_filelist.Add(filename, fci);
                }
            }
            int irows = 0;
            int batchsize = 200;//每批记录数
            List<object[]> dataObj = new List<object[]>();
            System.Threading.Tasks.Task taskitem = null;
            if (dft.Equals(DataFileType.DBF))//dbf
            {
                DbfFile odbf = new DbfFile(DataFuns.DefaultEncoding);
                try
                {
                    if (append)
                        odbf.Open(filename, FileMode.Open);
                    else
                        odbf.Create(filename);
                    DataTable table_info = GetSchemaTable(db_source, dtread);
                    for (int i = 0; i < dtread.FieldCount; i++)
                    {
                        string column_name = dtread.GetName(i);
                        string column_name_new = bcm.GetDestinationColumnName(i, column_name);
                        if (column_name_new.Length > 0)
                        {
                            Type fieldType = dtread.GetFieldType(i);
                            int max_length = 0;
                            int nLength = 0;
                            int nDecimals = 0;
                            DataRow[] drs = null;
                            drs = table_info.Select("ColumnName='" + column_name + "'");
                            if (drs.Length > 0)
                            {
                                int.TryParse(drs[0]["ColumnSize"].ToString(), out max_length);
                                if (max_length > 255) max_length = 255;
                                int.TryParse(drs[0]["NumericPrecision"].ToString(), out nLength);
                                if (nLength >= 255) nLength = 0;
                                int.TryParse(drs[0]["NumericScale"].ToString(), out nDecimals);
                                if (nDecimals >= 255) nDecimals = 0;
                                else if (nDecimals >= nLength) nDecimals = 2;
                            }
                            if (!append)
                            {
                                if (fieldType.IsValueType)
                                    odbf.Header.AddColumn(odbf.GetNewColumn(column_name_new, fieldType, nLength, nDecimals));
                                else
                                    odbf.Header.AddColumn(odbf.GetNewColumn(column_name_new, fieldType, max_length, 0));
                            }
                            fieldList.Add(i);
                        }
                    }
                    if (exp_filelist.ContainsKey(filename)) exp_filelist[filename].Position = odbf.Header.HeaderLength;
                    DbfRecord orec = new DbfRecord(odbf.Header) { AllowDecimalTruncate = true };
                    while (dtread.Read())//读取数据
                    {
                        object[] drdata = new object[fieldList.Count];
                        for (int i = 0; i < fieldList.Count; i++)
                        {
                            drdata[i] = dtread.GetValue(fieldList[i]);
                        }
                        dataObj.Add(drdata);//加入缓存
                        irows++;
                        if (irows == batchsize)
                        {
                            if (taskitem != null) taskitem.Wait();//等待线程写完
                            List<object[]> dataObj2 = dataObj;
                            taskitem = System.Threading.Tasks.Task.Run(() => StreamWriterDBFFileThread(odbf, orec, dataObj2));//启动写线程
                            dataObj = new List<object[]>();
                            irows = 0;
                        }
                    }
                    if (irows > 0)
                    {
                        if (taskitem != null) taskitem.Wait();//等待线程写完
                        List<object[]> dataObj2 = dataObj;
                        taskitem = System.Threading.Tasks.Task.Run(() => StreamWriterDBFFileThread(odbf, orec, dataObj2));//启动写线程
                    }
                    if (taskitem != null) taskitem.Wait();//等待线程写完
                    if (exp_filelist.ContainsKey(filename)) exp_filelist[filename].RecordCount = odbf.Header.RecordCount;
                }
                finally
                {
                    odbf.Close();
                }
            }
            else if (dft.Equals(DataFileType.XLS))//xls
            {
                DataTable dt = GetTableTable(bcm, dtread);
                Glpt.Office.ExportExcel xls = new Glpt.Office.ExportExcel();
                xls.SetDataSource(dt);
                xls.BuildSheet();
                xls.Save(filename);
            }
            else if (dft.Equals(DataFileType.XML))//xml
            {
                string DATASET_NAME = "DataSet1";
                if (param.ContainsKey("DATASET_NAME")) DATASET_NAME = param["DATASET_NAME"];
                string TABLE_NAME = "Table1";
                if (param.ContainsKey("TABLE_NAME")) DATASET_NAME = param["TABLE_NAME"];
                List<string> fieldNameList = new List<string>();
                for (int i = 0; i < dtread.FieldCount; i++)
                {
                    string column_name = bcm.GetDestinationColumnName(i, dtread.GetName(i));
                    if (column_name.Length > 0)
                    {
                        fieldNameList.Add(column_name);
                        fieldList.Add(i);
                    }
                }
                Encoding encoding = Encoding.UTF8;
                if (param.ContainsKey("CODEPAGE") && param["CODEPAGE"].Length > 0)
                    encoding = CommFuns.GetEncoding(param["CODEPAGE"]);
                using (StreamWriter file = new StreamWriter(filename, false, encoding))
                {
                    file.WriteLine("<?xml version=\"1.0\" standalone=\"yes\"?>");
                    file.WriteLine("<" + DATASET_NAME + ">");
                    while (dtread.Read())//读取数据
                    {
                        object[] drdata = new object[fieldList.Count];
                        for (int i = 0; i < fieldList.Count; i++)
                        {
                            drdata[i] = dtread.GetValue(fieldList[i]);
                        }
                        dataObj.Add(drdata);//加入缓存
                        irows++;
                        if (irows == batchsize)
                        {
                            if (taskitem != null) taskitem.Wait();//等待线程写完
                            List<object[]> dataObj2 = dataObj;
                            taskitem = System.Threading.Tasks.Task.Run(() => StreamWriterXMLFileThread(file, dataObj2, TABLE_NAME, fieldNameList));//启动写线程
                            dataObj = new List<object[]>();
                            irows = 0;
                        }
                    }
                    if (irows > 0)
                    {
                        if (taskitem != null) taskitem.Wait();//等待线程写完
                        List<object[]> dataObj2 = dataObj;
                        taskitem = System.Threading.Tasks.Task.Run(() => StreamWriterXMLFileThread(file, dataObj2, TABLE_NAME, fieldNameList));//启动写线程
                    }
                    if (taskitem != null) taskitem.Wait();//等待线程写完
                    file.WriteLine("</" + DATASET_NAME + ">");
                }
            }
            else//bcp,csv,json
            {
                bool hastitle = false;
                string separator = "\t";
                bool iscsv = dft.Equals(DataFileType.CSV);
                if (iscsv)//csv
                {
                    if (is_first) hastitle = true;
                    separator = ",";
                }
                if (param.ContainsKey("FIELDTERMINATOR"))
                    separator = param["FIELDTERMINATOR"];
                if (is_first)
                {
                    if (param.ContainsKey("TITLEROW") && param["TITLEROW"].Length > 0)
                        hastitle = param["TITLEROW"].Equals("1");
                    else if (!bcm.MAP_MODE.Equals("1"))
                        hastitle = true;
                }
                Encoding encoding = DataFuns.DefaultEncoding;
                if (param.ContainsKey("CODEPAGE") && param["CODEPAGE"].Length > 0)
                    encoding = CommFuns.GetEncoding(param["CODEPAGE"]);
                if (is_sql_bcp)//调用BCP导出命令
                {
                    string ROWTERMINATOR = "";
                    if (param.ContainsKey("ROWTERMINATOR") && param["ROWTERMINATOR"].Length > 0)
                        ROWTERMINATOR = param["ROWTERMINATOR"].Replace("\\n", "0x0D").Replace("\\r", "0x0A");
                    string arguments = "\"" + execute_sql.Replace("\"", "\"\"") + "\" queryout \"" + filename + "\" -c -S\"" + db_source.DB_IP + "\" -U\"" + db_source.USER_NAME + "\" -P\"" + AbstractInterFace.ReplaceString(db_source.USER_PASSWORD, AbstractInterFace.PublicLeftChar, AbstractInterFace.PublicRightChar).Replace("\"", "\"\"") + "\" -d\"" + db_source.DB_NAME + "\"";
                    if (separator.Length > 0 && !separator.Equals("\t")) arguments += " -t\"" + separator + "\"";
                    if (ROWTERMINATOR.Length > 0 && !ROWTERMINATOR.Equals("\r\n") && !ROWTERMINATOR.Equals("0x0A0x0D")) arguments += " -r\"" + ROWTERMINATOR + "\"";
                    dtread.Close();
                    CommFuns.Execute("bcp", arguments);
                }
                else
                {
                    using (StreamWriter file = new StreamWriter(filename, append, encoding))
                    {
                        if (dft.Equals(DataFileType.JSON))//json
                        {
                            DataTable dt = GetTableTable(bcm, dtread);
                            file.WriteLine(JsonDataTable.TableToJsonString(dt));
                        }
                        else//bcp,csv
                        {
                            for (int i = 0; i < dtread.FieldCount; i++)
                            {
                                string column_name = bcm.GetDestinationColumnName(i, dtread.GetName(i));
                                if (column_name.Length > 0)
                                {
                                    if (hastitle)
                                    {
                                        if (fieldList.Count > 0)
                                            file.Write(separator);
                                        file.Write(column_name);
                                    }
                                    fieldList.Add(i);
                                }
                            }
                            if (hastitle && fieldList.Count > 0) file.WriteLine("");
                            while (dtread.Read())//读取数据
                            {
                                object[] drdata = new object[fieldList.Count];
                                for (int i = 0; i < fieldList.Count; i++)
                                {
                                    drdata[i] = dtread.GetValue(fieldList[i]);
                                }
                                dataObj.Add(drdata);//加入缓存
                                irows++;
                                if (irows == batchsize)
                                {
                                    if (taskitem != null) taskitem.Wait();//等待线程写完
                                    List<object[]> dataObj2 = dataObj;
                                    taskitem = System.Threading.Tasks.Task.Run(() => StreamWriterBCPFileThread(file, dataObj2, separator, iscsv));//启动写线程
                                    dataObj = new List<object[]>();
                                    irows = 0;
                                }
                            }
                            if (irows > 0)
                            {
                                if (taskitem != null) taskitem.Wait();//等待线程写完
                                List<object[]> dataObj2 = dataObj;
                                taskitem = System.Threading.Tasks.Task.Run(() => StreamWriterBCPFileThread(file, dataObj2, separator, iscsv));//启动写线程
                            }
                            if (taskitem != null) taskitem.Wait();//等待线程写完
                        }
                    }
                }
            }
        }
        private void StreamWriterDBFFileThread(DbfFile odbf, DbfRecord orec, List<object[]> dataObj)
        {
            foreach (object[] drdata in dataObj)
            {
                for (int i = 0; i < drdata.Length; i++)
                {
                    object obj = drdata[i];
                    if (obj != null & !obj.Equals(DBNull.Value))
                        orec[i] = obj.ToString();
                }
                odbf.Write(orec, true);
            }
        }
        private void StreamWriterXMLFileThread(StreamWriter file, List<object[]> dataObj, string TABLE_NAME, List<string> fieldNameList)
        {
            foreach (object[] drdata in dataObj)
            {
                file.WriteLine("<" + TABLE_NAME + ">");
                for (int i = 0; i < drdata.Length; i++)
                {
                    file.WriteLine("<" + fieldNameList[i] + ">" + drdata[i].ToString().Replace("&", "&amp;").Replace("<", "&lt;").Replace(">", "&gt;") + "</" + fieldNameList[i] + ">");
                }
                file.WriteLine("</" + TABLE_NAME + ">");
            }
        }

        private void StreamWriterBCPFileThread(StreamWriter file, List<object[]> dataObj, string separator, bool iscsv)
        {
            foreach (object[] drdata in dataObj)
            {
                for (int i = 0; i < drdata.Length; i++)
                {
                    if (i > 0)
                        file.Write(separator);
                    object obj = drdata[i];
                    if (obj != null && !obj.Equals(DBNull.Value))
                    {
                        string stmp = obj.ToString();
                        if (stmp.Length > 0)
                        {
                            if (obj is DateTime)
                                stmp = ((DateTime)obj).ToString("yyyy-MM-dd HH:mm:ss.fff");
                            if (separator.Length > 0)
                            {
                                if (iscsv && stmp.Contains(separator))
                                    stmp = "\"" + stmp.Replace("\"", "\"\"") + "\"";
                                else
                                    stmp = stmp.Replace(separator, " ");
                            }
                            file.Write(stmp);
                        }
                        else
                            file.Write("\0");
                    }
                }
                file.WriteLine("");
            }
        }

        private void AddTableColumns(DataTable srcdt, Dictionary<string, string> param, string filename_source)
        {
            int icol = 0;
            foreach (string key in param.Keys)
            {
                if (key.StartsWith(":"))
                {
                    DataColumn dc = null;
                    string column_name = key.Substring(1);
                    if (param[key].Equals("IDENTITY"))
                    {
                        dc = new DataColumn(column_name, typeof(int));
                        srcdt.Columns.Add(dc);
                        for (int i = 0; i < srcdt.Rows.Count; i++)
                            srcdt.Rows[i][column_name] = i + 1;
                    }
                    else if (param[key].Equals("@SOURCE_TABLE_NAME"))
                    {
                        dc = new DataColumn(column_name, typeof(string));
                        srcdt.Columns.Add(dc);
                        for (int i = 0; i < srcdt.Rows.Count; i++)
                            srcdt.Rows[i][column_name] = filename_source;
                    }
                    else
                    {
                        dc = new DataColumn(column_name, typeof(string));
                        dc.DefaultValue = param[key];
                        try
                        {
                            srcdt.Columns.Add(dc);
                        }
                        catch (Exception ex)
                        {
                            if (!ex.Message.Contains("SimpleContent"))
                                throw;
                        }
                    }
                    dc.SetOrdinal(icol++);
                }
            }
        }
        private static DataTable GetTableTable(ColumnMap bcm, IDataReader dtread)
        {
            List<int> fieldList = new List<int>();
            DataTable dt = new DataTable();
            for (int i = 0; i < dtread.FieldCount; i++)
            {
                string column_name = bcm.GetDestinationColumnName(i, dtread.GetName(i));
                if (column_name.Length > 0)
                {
                    dt.Columns.Add(column_name);
                    fieldList.Add(i);
                }
            }
            while (dtread.Read())
            {
                DataRow dr = dt.NewRow();
                for (int i = 0; i < fieldList.Count; i++)
                {
                    dr[i] = dtread.GetValue(fieldList[i]);
                }
                dt.Rows.Add(dr);
            }
            return dt;
        }
        /// <summary>
        /// 自动登录共享网络
        /// </summary>
        /// <param name="filename">文件路径名</param>
        /// <param name="username">用户名</param>
        /// <param name="password">密码</param>
        private static void AutoLoginNet(string filename, string username, string password)
        {
            if (filename.StartsWith(@"\\") && username.Trim().Length > 0)
            {
                DirectoryInfo dinfo = new FileInfo(filename).Directory;
                while (dinfo.Parent != null)
                {
                    dinfo = dinfo.Parent;
                }
                if (!dinfo.Exists)
                {
                    password = AbstractInterFace.ReplaceString(password, AbstractInterFace.PublicLeftChar, AbstractInterFace.PublicRightChar);
                    using (Glpt.Common.SharedTool tool = new Glpt.Common.SharedTool(username, password, dinfo.FullName))
                    {
                    }
                }
            }
        }
        private static DataFileType GetDataFileType(string filename, string SQL_EXP)
        {
            Dictionary<string, string> param = DataFuns.SplitStringValue(SQL_EXP, ',');
            DataFileType dft = DataFileType.DATABASE;
            if (param.ContainsKey("FILE"))
            {
                string filetype = param["FILE"];
                if (filetype.Trim().Length == 0)
                    dft = DataFileType.DBF;
                else
                    dft = (DataFileType)Enum.Parse(typeof(DataFileType), filetype);
            }
            else if (param.ContainsKey("DLL"))
                dft = DataFileType.DLL;
            else
            {
                string fileext = "";
                if (param.ContainsKey("FILE_NAME"))
                    fileext = Path.GetExtension(param["FILE_NAME"].ToLower());
                else
                    fileext = Path.GetExtension(filename.ToLower());
                if (fileext.Equals(".pdf"))//PDF文件
                    dft = DataFileType.PDF;
                else if (param.ContainsKey("PATTERN") || fileext.Equals(".htm") || fileext.Equals(".html"))//HTML文件
                    dft = DataFileType.HTML;
                else if (param.ContainsKey("FIELDTERMINATOR") || fileext.Equals(".txt") || fileext.Equals(".bcp") || fileext.Equals(".dat"))//BCP文本文件
                    dft = DataFileType.BCP;
                else if (fileext.Equals(".xls") || fileext.Equals(".xlsx"))//XLS文件
                    dft = DataFileType.XLS;
                else if (fileext.Equals(".csv") && !SQL_EXP.Trim().ToUpper().StartsWith("SELECT"))//CSV文件
                    dft = DataFileType.CSV;
                else if (fileext.Equals(".xml"))//XML文件
                    dft = DataFileType.XML;
                else if (fileext.Equals(".json"))//JSON文件
                    dft = DataFileType.JSON;
                else if (fileext.Equals(".dbf") && !SQL_EXP.Trim().ToUpper().StartsWith("SELECT") || CommFuns.IsDbf(fileext))//DBF文件
                    dft = DataFileType.DBF;
                else if (FilePath.IsURL(filename))
                    dft = DataFileType.HTML;//URL
                else if (fileext.Equals(".doc"))//DOC文件
                    dft = DataFileType.DOC;
            }
            return dft;
        }
        /// <summary>
        /// 获取数据库项目信息
        /// </summary>
        private DbItem GetDbItem(BaseInterFace bif, string DB_LIST, string DB_NO)
        {
            DbItem dbi = null;
            if (DB_LIST.Equals("0"))
            {
                dbi = BaseInterFace.DbList.GetByNo(int.Parse(DB_NO));
                if (dbi == null)
                    throw new Exception("没有找到数据库编号" + DB_NO);
            }
            else
            {
                string sql = "SELECT * FROM TB_EXCH_DATABASE WHERE DB_NO=" + DB_NO;
                DataTable dt_source = bif.ExecSelect(sql);
                if (dt_source.Rows.Count == 0)
                    throw new Exception("没有找到数据库编号" + DB_NO);
                else
                {
                    DataRow dr = dt_source.Rows[0];
                    dr["DB_IP"] = dr["DB_IP"].ToString().Trim();
                    dr["DB_NAME"] = dr["DB_NAME"].ToString().Trim();
                    dbi = new DbItem(dr);
                }
            }
            return dbi;
        }
        /// <summary>
        /// 获取数据库项目集群
        /// </summary>
        private DbItemCollection GetDbItemCollection(BaseInterFace bif, string DB_LIST, string DB_NO)
        {
            DbItemCollection dbic = new DbItemCollection();
            DbItem dbi = GetDbItem(bif, DB_LIST, DB_NO);
            if (dbi != null)
            {
                if (DB_LIST.Equals("0"))
                {
                    if (BaseInterFace.DbListCluster.ContainsKey(dbi))
                    {
                        dbic = BaseInterFace.DbListCluster[dbi];
                    }
                }
                else if (dbi.GROUP_NO > 0 && dbi.GROUP_NO == dbi.DB_NO)
                {
                    string sql = "SELECT * FROM TB_EXCH_DATABASE WHERE GROUP_NO=" + dbi.GROUP_NO.ToString();
                    DataTable dt_source = bif.ExecSelect(sql);
                    foreach (DataRow dr in dt_source.Rows)
                    {
                        dr["DB_IP"] = dr["DB_IP"].ToString().Trim();
                        dr["DB_NAME"] = dr["DB_NAME"].ToString().Trim();
                        dbic.Add(new DbItem(dr));
                    }
                }
                if (dbic.Count == 0) dbic.Add(dbi);
            }
            return dbic;
        }

        private List<DataTaskParam> SortRandom(List<DataTaskParam> plist)
        {
            if (plist.Count <= 1) return plist;
            List<DataTaskParam> ordlist = new List<DataTaskParam>();
            ordlist.Add(plist[0]);
            int[] rand = CommFuns.GetRandomArray(plist.Count - 1);//产生随机序号
            foreach (int id in rand)
            {
                ordlist.Add(plist[id + 1]);
            }
            return ordlist;
        }

        private string ReplaceNetPath(string path)
        {
            if (!IsWeb)
                return path;
            foreach (DataRow dr in NetPathMapTable.Rows)
            {
                string NET_PATH = dr["NET_PATH"].ToString();
                if (path.StartsWith(NET_PATH))
                {
                    path = path.Replace(NET_PATH, dr["MAP_PATH"].ToString());
                    break;
                }
            }
            return path;
        }
        private object ConvertToDateTime(string val)
        {
            int len = val.Length;
            if (len > 0)
            {
                string format = "yyyyMMdd";
                if (len == 10) format = "yyyy-MM-dd";
                else if (len == 14) format = "yyyyMMddHHmmss";
                else if (len == 19) format = "yyyy-MM-dd HH:mm:ss";
                else if (len == 23) format = "yyyy-MM-dd HH:mm:ss.fff";
                return DateTime.ParseExact(val, format, System.Globalization.CultureInfo.InvariantCulture);
            }
            else
                return DBNull.Value;
        }
        public static bool GetParamDictionary(string task_id, Dictionary<string, string> param, ref int BATCHSIZE, ref int THREADCOUNT, ref int WRITE_THREADCOUNT, ref int TIMEOUT, ref int MAXERRORS, ref bool IsFileLog, ref string SPLIT_COLNAME)
        {
            bool ContainsKey = false;
            if (param.ContainsKey("ALLOWNOTEXIST"))
            {
                ContainsKey = true;
            }
            if (param.ContainsKey("BATCHSIZE"))
            {
                BATCHSIZE = int.Parse(param["BATCHSIZE"]);
                ContainsKey = true;
            }
            if (param.ContainsKey("THREADCOUNT"))
            {
                THREADCOUNT = int.Parse(param["THREADCOUNT"]);
                ContainsKey = true;
            }
            if (param.ContainsKey("WRITE_THREADCOUNT"))
            {
                WRITE_THREADCOUNT = int.Parse(param["WRITE_THREADCOUNT"]);
                ContainsKey = true;
            }
            if (param.ContainsKey("TIMEOUT"))
            {
                TIMEOUT = int.Parse(param["TIMEOUT"]);
                ContainsKey = true;
            }
            if (param.ContainsKey("MAXERRORS"))
            {
                MAXERRORS = int.Parse(param["MAXERRORS"]);
                ContainsKey = true;
            }
            if (param.ContainsKey("DEBUG"))
            {
                if (task_id.Length > 0) IsFileLog = param["DEBUG"].Equals("1");
                ContainsKey = true;
            }
            if (param.ContainsKey("NOT_UNION"))
            {
                ContainsKey = true;
            }
            if (param.ContainsKey("SPLIT_COLNAME"))
            {
                SPLIT_COLNAME = param["SPLIT_COLNAME"];
                ContainsKey = true;
            }
            return ContainsKey;
        }

        /// <summary>
        /// 后台任务对象
        /// </summary>
        public BackgroundWorker BackgroundWorkerTask
        {
            get
            {
                return bwt;
            }
            set
            {
                bwt = value;
            }
        }

        /// <summary>
        /// 释放内存
        /// </summary>
        public void Dispose()
        {
            try
            {
                GC.Collect();
            }
            catch { }
        }

        /// <summary>
        /// 析构函数
        /// </summary>
        ~ExchDataTask()
        {
            Dispose();
        }
    }

    public class DataTaskParam
    {
        public string BUSI_DATE;
        public DbItem DB_SOURCE;
        public DbItem DB_TARGET;
        public bool IMP_FILE;
        public List<string> FILENAME_SOURCE_LIST;
        public string FILENAME_SOURCE;
        public string SOURCE_TABLE_NAME;
        public string TARGET_TABLE_NAME;
        public string SQL;
        public string SQL_EXP;
        public string OP_EXP;
        public List<string> OP_EXP_LIST;
        public string TRAN_MODE;
        public string MAP_PARAMS;
        public bool ALLOWNOTEXIST;
        public int BATCHSIZE;
        public int THREADCOUNT;
    }

    public class ColumnMapInfo
    {
        public int ColumnId;
        public string ColumnName;
        public int ColumnOrdinal;
        public int ColumnSize;
        public Type FieldType;
    }

    public class ExecuteInfo
    {
        public string TEMP_TABLE_NAME = "";
        public int EXECUTE_COUNT = 0;
        public List<string> DELETE_SQL = new List<string>();
    }

    public class FileCombineInfo
    {
        public DataFileType FileType = DataFileType.BCP;
        public string FileName = "";
        public long Position = 0;
        public uint RecordCount = 0;
    }
}
